diff --git a/README.md b/README.md index e77ea25..9e9282c 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,7 @@ cd packages/api && bun run scripts/service-matrix.ts | Integration | API Gateway | Yes (list, create, delete, inspect) | No | No | | Integration | SNS / Notifications | Yes (list, create, inspect, delete) | No | No | | Integration | EventBridge / Events | Yes (list, create, delete, inspect) | No | No | +| Integration | EventBridge Pipes / Pipes | Yes (list, create, delete, inspect) | No | No | | Integration | SES Mailbox / Email | Yes (list, inspect) | No | No | | Integration | Cloud Scheduler | No | No | Yes (list, create, delete, inspect) | | Integration | Step Functions / Workflows | Yes (list, create, update, delete, inspect) | No | No | diff --git a/packages/api/package.json b/packages/api/package.json index 9228d93..798f994 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -36,6 +36,7 @@ "@aws-sdk/client-kms": "^3.1096.0", "@aws-sdk/client-lambda": "^3.1076.0", "@aws-sdk/client-lightsail": "^3.1076.0", + "@aws-sdk/client-pipes": "^3.1146.0", "@aws-sdk/client-rds": "^3.1076.0", "@aws-sdk/client-rds-data": "^3.1076.0", "@aws-sdk/client-route-53": "^3.1146.0", diff --git a/packages/api/src/adapter-aws/AwsPipesAdapter.test.ts b/packages/api/src/adapter-aws/AwsPipesAdapter.test.ts new file mode 100644 index 0000000..ecf91c0 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsPipesAdapter.test.ts @@ -0,0 +1,122 @@ +import {describe, expect, test} from 'bun:test' +import { + CreatePipeCommand, + DeletePipeCommand, + DescribePipeCommand, + ListPipesCommand, + type PipesClient, + StartPipeCommand, + StopPipeCommand, +} from '@aws-sdk/client-pipes' +import {AwsPipesAdapter, describeArn} from './AwsPipesAdapter' +import {ValidationError} from '../cloud-spi/errors' + +type SendResult = Record + +function stubPipes(handler: (command: object) => SendResult | Promise) { + const sent: object[] = [] + const client = { + async send(command: object) { + sent.push(command) + return handler(command) + }, + } as unknown as PipesClient + return {client, sent, adapter: new AwsPipesAdapter(client)} +} + +const SRC = 'arn:aws:sqs:us-east-1:000000000000:pipe-src' +const DST = 'arn:aws:lambda:us-east-1:000000000000:function:worker' +const ROLE = 'arn:aws:iam::000000000000:role/pipe-role' +const described = { + Name: 'orders-pipe', Arn: 'arn:aws:pipes:us-east-1:000000000000:pipe/orders-pipe', + DesiredState: 'RUNNING', CurrentState: 'RUNNING', Source: SRC, Target: DST, RoleArn: ROLE, + SourceParameters: {FilterCriteria: {Filters: [{Pattern: '{"body":{"kind":["order"]}}'}]}, SqsQueueParameters: {BatchSize: 5}}, + TargetParameters: {InputTemplate: '{"id": <$.messageId>}'}, + CreationTime: new Date('2026-10-25T16:42:45Z'), +} + +describe('AwsPipesAdapter', () => { + test('lists pipes across pages with readable source and target names', async () => { + const {adapter} = stubPipes((command) => { + if (!(command instanceof ListPipesCommand)) return {} + return command.input.NextToken + ? {Pipes: [{Name: 'clicks', CurrentState: 'STOPPED', Source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', Target: 'arn:aws:events:us-east-1:000000000000:event-bus/default'}]} + : {Pipes: [{Name: 'orders-pipe', CurrentState: 'RUNNING', Source: SRC, Target: DST}], NextToken: 'n'} + }) + + const pipes = await adapter.list() + expect(pipes.map((pipe) => [pipe.name, pipe.status, pipe.metadata.sourceName, pipe.metadata.targetName])).toEqual([ + ['orders-pipe', 'RUNNING', 'pipe-src (SQS queue)', 'worker (Lambda function)'], + ['clicks', 'STOPPED', 'clicks (Kinesis stream)', 'default (EventBridge event bus)'], + ]) + expect((await adapter.list({search: 'kinesis'})).map((pipe) => pipe.name)).toEqual(['clicks']) + }) + + test('describes the filter, batch size, template and role', async () => { + const {adapter} = stubPipes(() => described) + expect((await adapter.get('orders-pipe'))?.metadata).toMatchObject({ + filterPatterns: ['{"body":{"kind":["order"]}}'], + batchSize: 5, + inputTemplate: '{"id": <$.messageId>}', + roleArn: ROLE, + }) + }) + + test('a missing pipe is null', async () => { + const {adapter} = stubPipes(() => { throw Object.assign(new Error('gone'), {name: 'NotFoundException'}) }) + expect(await adapter.get('nope')).toBeNull() + }) + + test('creates an SQS pipe with its filter, batch size and template', async () => { + const {adapter, sent} = stubPipes((command) => command instanceof DescribePipeCommand ? described : {}) + + await adapter.create({values: {name: 'orders-pipe', source: SRC, target: DST, roleArn: ROLE, batchSize: '5', filterPattern: '{ "body": { "kind": ["order"] } }', inputTemplate: '{"id": <$.messageId>}'}}) + expect((sent[0] as CreatePipeCommand).input).toEqual({ + Name: 'orders-pipe', + Source: SRC, + SourceParameters: {FilterCriteria: {Filters: [{Pattern: '{"body":{"kind":["order"]}}'}]}, SqsQueueParameters: {BatchSize: 5}}, + Target: DST, + TargetParameters: {InputTemplate: '{"id": <$.messageId>}'}, + RoleArn: ROLE, + DesiredState: 'RUNNING', + }) + }) + + test('gives stream and MSK sources the starting position they require', async () => { + const {adapter, sent} = stubPipes((command) => command instanceof DescribePipeCommand ? described : {}) + + await adapter.create({values: {name: 'clicks', source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', target: DST, roleArn: ROLE, startingPosition: 'TRIM_HORIZON', desiredState: 'STOPPED'}}) + expect((sent[0] as CreatePipeCommand).input).toMatchObject({SourceParameters: {KinesisStreamParameters: {StartingPosition: 'TRIM_HORIZON'}}, DesiredState: 'STOPPED'}) + + await adapter.create({values: {name: 'k', source: 'arn:aws:kafka:us-east-1:000000000000:cluster/events/abc', target: DST, roleArn: ROLE, topicName: 'orders'}}) + expect((sent[2] as CreatePipeCommand).input.SourceParameters).toEqual({ManagedStreamingKafkaParameters: {TopicName: 'orders', StartingPosition: 'LATEST'}}) + }) + + test('refuses pipes the runtime could not run, before calling it', async () => { + const {adapter, sent} = stubPipes(() => ({})) + const base = {name: 'p', source: SRC, target: DST, roleArn: ROLE} + await expect(adapter.create({values: {...base, target: 'arn:aws:s3:::bucket'}})).rejects.toThrow('target must be the ARN of a') + await expect(adapter.create({values: {...base, source: 'arn:aws:sns:us-east-1:000000000000:topic'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, enrichment: 'arn:aws:sqs:us-east-1:000000000000:q'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, filterPattern: '{kind: order}'}})).rejects.toThrow('filterPattern must be JSON') + await expect(adapter.create({values: {...base, batchSize: '20000'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, source: 'arn:aws:kafka:us-east-1:000000000000:cluster/c/1'}})).rejects.toThrow('topicName is required') + await expect(adapter.create({values: {...base, roleArn: 'pipe-role'}})).rejects.toBeInstanceOf(ValidationError) + expect(sent).toHaveLength(0) + }) + + test('starts, stops and deletes by name', async () => { + const {adapter, sent} = stubPipes(() => ({})) + await adapter.stop('orders-pipe') + await adapter.start('orders-pipe') + await adapter.delete('orders-pipe') + expect(sent.map((command) => command.constructor)).toEqual([StopPipeCommand, StartPipeCommand, DeletePipeCommand]) + expect(sent.every((command) => (command as StopPipeCommand).input.Name === 'orders-pipe')).toBe(true) + }) + + test('names every kind of source and target ARN', () => { + expect(describeArn('arn:aws:dynamodb:us-east-1:000000000000:table/orders/stream/2026-10-25T00:00:00.000')).toBe('orders (DynamoDB stream)') + expect(describeArn('arn:aws:states:us-east-1:000000000000:stateMachine:flow')).toBe('flow (Step Functions state machine)') + expect(describeArn('arn:aws:sns:us-east-1:000000000000:alerts')).toBe('alerts (SNS topic)') + }) +}) diff --git a/packages/api/src/adapter-aws/AwsPipesAdapter.ts b/packages/api/src/adapter-aws/AwsPipesAdapter.ts new file mode 100644 index 0000000..4dd23a4 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsPipesAdapter.ts @@ -0,0 +1,237 @@ +import { + CreatePipeCommand, + DeletePipeCommand, + DescribePipeCommand, + type DescribePipeCommandOutput, + ListPipesCommand, + type Pipe, + type PipeSourceParameters, + type PipesClient, + StartPipeCommand, + StopPipeCommand, +} from '@aws-sdk/client-pipes' +import {RuntimeError, ValidationError} from '../cloud-spi/errors' +import { + awsPipesSchema, + PIPE_ENRICHMENT_SERVICES, + PIPE_NAME_PATTERN, + PIPE_SOURCE_SERVICES, + PIPE_STARTING_POSITIONS, + PIPE_TARGET_SERVICES, +} from '../cloud-spi/pipesSchema' +import type {CloudResource, CloudServiceAdapter, CreateResourceInput, ResourceQuery, ServiceSchema} from '../cloud-spi/types' + +const MAX_BATCH_SIZE = 10_000 + +/** + * Talks to EventBridge Pipes, verified against Floci core (community, `latest`). + * + * Notes from probing the runtime: + * - Pipes really run: the runtime polls SQS, Kinesis, DynamoDB streams and MSK, + * applies the filter, enrichment and input template, and calls the target. + * - Its poller delivers only to Lambda, SQS, SNS, EventBridge buses and Step + * Functions, so create refuses other targets instead of making a pipe that + * would never deliver. + * - ListPipes returns no parameters, so the table shows only what it carries; + * the filter, template and role come from DescribePipe. + */ +export class AwsPipesAdapter implements CloudServiceAdapter { + readonly cloud = 'aws' as const + readonly service = 'pipes' as const + + constructor(private readonly pipes: PipesClient) {} + + schema(): ServiceSchema { + return awsPipesSchema() + } + + async list(query: ResourceQuery = {}): Promise { + const found: Pipe[] = [] + let next: string | undefined + do { + const res = await this.pipes.send(new ListPipesCommand({Limit: 100, ...(next ? {NextToken: next} : {})})) + found.push(...(res.Pipes ?? [])) + next = res.NextToken + } while (next) + + const search = query.search?.trim().toLowerCase() + return found.map((pipe) => toResource(pipe)).filter((pipe) => !search + || pipe.name.toLowerCase().includes(search) + || String(pipe.metadata.source).toLowerCase().includes(search) + || String(pipe.metadata.target).toLowerCase().includes(search)) + } + + async get(id: string): Promise { + try { + return toResource(await this.pipes.send(new DescribePipeCommand({Name: id}))) + } catch (error) { + if (errorName(error) === 'NotFoundException') return null + throw error + } + } + + async create(input: CreateResourceInput): Promise { + const values = input.values + const name = text(values.name) + if (!new RegExp(PIPE_NAME_PATTERN).test(name)) throw new ValidationError('name must be 1 to 64 letters, numbers, dots, dashes or underscores') + + const source = text(values.source) + const sourceService = supported(source, PIPE_SOURCE_SERVICES, 'source') + const target = text(values.target) + supported(target, PIPE_TARGET_SERVICES, 'target') + const enrichment = text(values.enrichment) + if (enrichment) supported(enrichment, PIPE_ENRICHMENT_SERVICES, 'enrichment') + const roleArn = text(values.roleArn) + if (!/^arn:aws[a-z-]*:iam::\d{12}:role\/.+$/.test(roleArn)) throw new ValidationError('roleArn must be an IAM role ARN') + + const inputTemplate = text(values.inputTemplate) + const description = text(values.description) + const desiredState = values.desiredState === 'STOPPED' ? 'STOPPED' : 'RUNNING' + + await this.pipes.send(new CreatePipeCommand({ + Name: name, + Source: source, + SourceParameters: sourceParameters(sourceService, values), + Target: target, + ...(inputTemplate ? {TargetParameters: {InputTemplate: inputTemplate}} : {}), + ...(enrichment ? {Enrichment: enrichment} : {}), + RoleArn: roleArn, + DesiredState: desiredState, + ...(description ? {Description: description} : {}), + })) + + const created = await this.get(name) + if (!created) throw new RuntimeError(`EventBridge Pipes did not return the new pipe ${name}`) + return created + } + + async delete(id: string): Promise { + await this.pipes.send(new DeletePipeCommand({Name: id})) + } + + async start(id: string): Promise { + await this.pipes.send(new StartPipeCommand({Name: id})) + } + + async stop(id: string): Promise { + await this.pipes.send(new StopPipeCommand({Name: id})) + } +} + +/** Each source kind carries its own parameter block, as CreatePipe expects. */ +function sourceParameters(service: keyof typeof PIPE_SOURCE_SERVICES, values: Record): PipeSourceParameters { + const batchSize = optionalBatchSize(values.batchSize) + const filterPattern = text(values.filterPattern) + const parameters: PipeSourceParameters = filterPattern ? {FilterCriteria: {Filters: [{Pattern: jsonObject(filterPattern, 'filterPattern')}]}} : {} + const batch = batchSize ? {BatchSize: batchSize} : {} + + if (service === 'sqs') return {...parameters, SqsQueueParameters: batch} + + const startingPosition = text(values.startingPosition) || 'LATEST' + if (!(PIPE_STARTING_POSITIONS as readonly string[]).includes(startingPosition)) { + throw new ValidationError(`startingPosition must be one of ${PIPE_STARTING_POSITIONS.join(', ')}`) + } + const position = startingPosition as (typeof PIPE_STARTING_POSITIONS)[number] + if (service === 'kinesis') return {...parameters, KinesisStreamParameters: {StartingPosition: position, ...batch}} + if (service === 'dynamodb') return {...parameters, DynamoDBStreamParameters: {StartingPosition: position, ...batch}} + + const topicName = text(values.topicName) + if (!topicName) throw new ValidationError('topicName is required for an MSK source') + return {...parameters, ManagedStreamingKafkaParameters: {TopicName: topicName, StartingPosition: position, ...batch}} +} + +function toResource(pipe: Pipe | DescribePipeCommandOutput): CloudResource { + const name = pipe.Name ?? '' + const detail = 'SourceParameters' in pipe ? pipe : undefined + const sourceParameters = detail?.SourceParameters + const batchSize = sourceParameters?.SqsQueueParameters?.BatchSize + ?? sourceParameters?.KinesisStreamParameters?.BatchSize + ?? sourceParameters?.DynamoDBStreamParameters?.BatchSize + ?? sourceParameters?.ManagedStreamingKafkaParameters?.BatchSize + return { + id: name, + name, + cloud: 'aws', + service: 'pipes', + type: 'pipe', + region: pipe.Arn?.split(':')[3] || null, + createdAt: pipe.CreationTime ? pipe.CreationTime.toISOString() : null, + status: pipe.CurrentState ?? null, + metadata: { + arn: pipe.Arn, + desiredState: pipe.DesiredState ?? null, + currentState: pipe.CurrentState ?? null, + stateReason: pipe.StateReason ?? null, + source: pipe.Source ?? null, + sourceName: describeArn(pipe.Source), + target: pipe.Target ?? null, + targetName: describeArn(pipe.Target), + enrichment: pipe.Enrichment ?? null, + lastModified: pipe.LastModifiedTime ? pipe.LastModifiedTime.toISOString() : null, + ...(detail ? { + description: detail.Description ?? null, + roleArn: detail.RoleArn ?? null, + filterPatterns: (sourceParameters?.FilterCriteria?.Filters ?? []).map((filter) => filter.Pattern ?? '').filter(Boolean), + batchSize: batchSize ?? null, + startingPosition: sourceParameters?.KinesisStreamParameters?.StartingPosition + ?? sourceParameters?.DynamoDBStreamParameters?.StartingPosition + ?? sourceParameters?.ManagedStreamingKafkaParameters?.StartingPosition + ?? null, + topicName: sourceParameters?.ManagedStreamingKafkaParameters?.TopicName ?? null, + inputTemplate: detail.TargetParameters?.InputTemplate ?? null, + enrichmentInputTemplate: detail.EnrichmentParameters?.InputTemplate ?? null, + } : {}), + }, + } +} + +/** + * A readable name for an ARN: `queue (SQS queue)`. The resource part differs by + * service — `function:name`, `stream/name`, `table/name/stream/…`, `cluster/name/uuid`. + */ +export function describeArn(arn: string | undefined): string | null { + if (!arn) return null + const parts = arn.split(':') + const service = parts[2] ?? '' + const resource = parts.slice(5).join(':') + const name = service === 'lambda' || service === 'states' ? resource.split(':')[1] ?? resource + : service === 'dynamodb' || service === 'kafka' || service === 'kinesis' || service === 'events' ? resource.split('/')[1] ?? resource + : resource + const kind = (PIPE_SOURCE_SERVICES as Record)[service] ?? (PIPE_TARGET_SERVICES as Record)[service] + return kind ? `${name} (${kind})` : name +} + +function supported>(arn: string, services: T, field: string): keyof T { + const service = arn.split(':')[2] ?? '' + if (!arn.startsWith('arn:') || !(service in services)) { + throw new ValidationError(`${field} must be the ARN of a ${Object.values(services).join(', ')}`) + } + return service as keyof T +} + +function optionalBatchSize(value: unknown): number | undefined { + const raw = text(value) + if (!raw) return undefined + const size = Number(raw) + if (!Number.isInteger(size) || size < 1 || size > MAX_BATCH_SIZE) throw new ValidationError(`batchSize must be a whole number from 1 to ${MAX_BATCH_SIZE}`) + return size +} + +function jsonObject(value: string, field: string): string { + let parsed: unknown + try { + parsed = JSON.parse(value) + } catch { + throw new ValidationError(`${field} must be JSON, such as {"body": {"kind": ["order"]}}`) + } + if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) throw new ValidationError(`${field} must be a JSON object`) + return JSON.stringify(parsed) +} + +function text(value: unknown): string { + return typeof value === 'string' ? value.trim() : '' +} + +function errorName(error: unknown): string | undefined { + return typeof error === 'object' && error !== null ? (error as {name?: string}).name : undefined +} diff --git a/packages/api/src/aws.ts b/packages/api/src/aws.ts index 86a2b13..4074777 100644 --- a/packages/api/src/aws.ts +++ b/packages/api/src/aws.ts @@ -34,6 +34,7 @@ import { LightsailClient } from "@aws-sdk/client-lightsail"; import { CognitoIdentityProviderClient } from "@aws-sdk/client-cognito-identity-provider"; import { Route53Client } from "@aws-sdk/client-route-53"; import { CloudFrontClient } from "@aws-sdk/client-cloudfront"; +import { PipesClient } from "@aws-sdk/client-pipes"; const endpoint = process.env.FLOCI_ENDPOINT; const region = process.env.AWS_REGION || "us-east-1"; @@ -99,6 +100,7 @@ export type AwsClients = { cognito: CognitoIdentityProviderClient; route53: Route53Client; cloudfront: CloudFrontClient; + pipes: PipesClient; }; export type AwsClientName = keyof AwsClients; @@ -148,6 +150,7 @@ function buildClients(accountId: string): AwsClients { cognito: new CognitoIdentityProviderClient(base), route53: new Route53Client(base), cloudfront: new CloudFrontClient(base), + pipes: new PipesClient(base), }; } diff --git a/packages/api/src/cloud-spi/pipesSchema.ts b/packages/api/src/cloud-spi/pipesSchema.ts new file mode 100644 index 0000000..daa7e46 --- /dev/null +++ b/packages/api/src/cloud-spi/pipesSchema.ts @@ -0,0 +1,151 @@ +import type {CloudProvider, FieldSchema, ServiceSchema, TableColumnSchema} from './types' + +/** CreatePipe's name rule. */ +export const PIPE_NAME_PATTERN = '^[\\.\\-_A-Za-z0-9]{1,64}$' +export const PIPE_STARTING_POSITIONS = ['TRIM_HORIZON', 'LATEST'] as const + +/** + * The sources and targets the runtime's pipe poller actually moves events + * between, keyed by the ARN's service field. Anything else is refused on create + * rather than accepted into a pipe that would never deliver. + */ +export const PIPE_SOURCE_SERVICES = {sqs: 'SQS queue', kinesis: 'Kinesis stream', dynamodb: 'DynamoDB stream', kafka: 'MSK topic'} as const +export const PIPE_TARGET_SERVICES = {lambda: 'Lambda function', sqs: 'SQS queue', sns: 'SNS topic', events: 'EventBridge event bus', states: 'Step Functions state machine'} as const +export const PIPE_ENRICHMENT_SERVICES = {lambda: 'Lambda function', states: 'Step Functions state machine'} as const + +const ARN = '^arn:aws[a-z-]*:[a-z0-9-]+:[a-z0-9-]*:\\d{12}:.+$' + +const pipeColumns: TableColumnSchema[] = [ + {name: 'name', label: 'Name'}, + {name: 'status', label: 'State', format: 'badge'}, + {name: 'source', label: 'Source', path: 'metadata.sourceName'}, + {name: 'target', label: 'Target', path: 'metadata.targetName'}, + {name: 'createdAt', label: 'Created', format: 'datetime'}, +] + +const pipeFilters: FieldSchema[] = [{name: 'search', label: 'Search', type: 'text', required: false}] + +export function awsPipesSchema(): ServiceSchema { + return { + cloud: 'aws', + service: 'pipes', + displayName: 'EventBridge Pipes', + fields: [ + { + name: 'name', + label: 'Name', + type: 'text', + required: true, + validation: {pattern: PIPE_NAME_PATTERN, message: 'Use up to 64 letters, numbers, dots, dashes and underscores'}, + }, + {name: 'description', label: 'Description', type: 'text', required: false, validation: {maxLength: 512}}, + { + name: 'source', + label: 'Source ARN', + type: 'text', + required: true, + span: true, + description: 'Where events come from: an SQS queue, a Kinesis stream, a DynamoDB stream or an MSK cluster.', + validation: {pattern: ARN, message: 'Use the ARN of a queue or stream'}, + }, + { + name: 'startingPosition', + label: 'Starting position', + type: 'select', + required: false, + defaultValue: 'LATEST', + description: 'For streams and MSK: read new records only, or everything still in the stream.', + options: [ + {label: 'New records only (LATEST)', value: 'LATEST'}, + {label: 'Oldest record (TRIM_HORIZON)', value: 'TRIM_HORIZON'}, + ], + }, + { + name: 'topicName', + label: 'MSK topic', + type: 'text', + required: false, + description: 'Required when the source is an MSK cluster.', + }, + { + name: 'batchSize', + label: 'Batch size', + type: 'text', + required: false, + description: 'Records sent to the target at once. 1 to 10,000; leave empty for the default of 10.', + validation: {pattern: '^[1-9][0-9]{0,4}$', message: 'Use a whole number from 1 to 10000'}, + }, + { + name: 'filterPattern', + label: 'Filter pattern', + type: 'textarea', + required: false, + span: true, + description: 'Optional EventBridge event pattern; only matching records go on. For SQS, match the parsed message under body, for example {"body": {"kind": ["order"]}}.', + }, + { + name: 'enrichment', + label: 'Enrichment ARN', + type: 'text', + required: false, + span: true, + description: 'Optional Lambda function or Step Functions state machine that transforms each batch before the target gets it.', + validation: {pattern: ARN, message: 'Use the ARN of a Lambda function or state machine'}, + }, + { + name: 'target', + label: 'Target ARN', + type: 'text', + required: true, + span: true, + description: 'Where events go: a Lambda function, SQS queue, SNS topic, EventBridge event bus or Step Functions state machine.', + validation: {pattern: ARN, message: 'Use the ARN of the target'}, + }, + { + name: 'inputTemplate', + label: 'Target input template', + type: 'textarea', + required: false, + span: true, + description: 'Optional. Reshape what the target receives, with <$.path> placeholders, for example {"id": <$.messageId>}.', + }, + { + name: 'roleArn', + label: 'Execution role ARN', + type: 'text', + required: true, + span: true, + description: 'The IAM role the pipe uses to read the source and call the target, as AWS requires. Create one in IAM first.', + validation: {pattern: '^arn:aws[a-z-]*:iam::\\d{12}:role/.+$', message: 'Use an IAM role ARN such as arn:aws:iam::000000000000:role/pipe-role'}, + }, + { + name: 'desiredState', + label: 'State after creating', + type: 'select', + required: false, + defaultValue: 'RUNNING', + options: [ + {label: 'Running', value: 'RUNNING'}, + {label: 'Stopped', value: 'STOPPED'}, + ], + }, + ], + actions: ['list', 'create', 'delete', 'inspect'], + capabilities: { + resourceActions: [ + {name: 'list', label: 'List pipes', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'create', label: 'Create pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'delete', label: 'Delete pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'inspect', label: 'Inspect pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'start', label: 'Start pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'stop', label: 'Stop pipe', enabled: true, status: 'available', runtimeRequired: true}, + ], + }, + filters: pipeFilters, + columns: pipeColumns, + } +} + +export function pipesSchemaFor(cloud: CloudProvider): ServiceSchema | null { + return cloud === 'aws' ? awsPipesSchema() : null +} diff --git a/packages/api/src/cloud-spi/serviceCatalog.ts b/packages/api/src/cloud-spi/serviceCatalog.ts index c297a41..dcedb16 100644 --- a/packages/api/src/cloud-spi/serviceCatalog.ts +++ b/packages/api/src/cloud-spi/serviceCatalog.ts @@ -274,6 +274,13 @@ export const SERVICE_CATALOG = { group: 'Provisioning', order: 20, }, + pipes: { + displayName: 'Pipes', + displayNameByCloud: {aws: 'EventBridge Pipes'}, + iconKey: 'pipes', + group: 'Integration', + order: 25, + }, scheduler: { displayName: 'Cloud Scheduler', displayNameByCloud: {gcp: 'Cloud Scheduler'}, diff --git a/packages/api/src/cloudProxy.ts b/packages/api/src/cloudProxy.ts index 48ebcaf..6405a7f 100644 --- a/packages/api/src/cloudProxy.ts +++ b/packages/api/src/cloudProxy.ts @@ -50,6 +50,7 @@ import {AwsLightsailAdapter} from './adapter-aws/AwsLightsailAdapter' import {AwsUserPoolsAdapter} from './adapter-aws/AwsUserPoolsAdapter' import {AwsDnsAdapter} from './adapter-aws/AwsDnsAdapter' import {AwsCdnAdapter} from './adapter-aws/AwsCdnAdapter' +import {AwsPipesAdapter} from './adapter-aws/AwsPipesAdapter' import {AwsMonitoringAdapter} from './adapter-aws/AwsMonitoringAdapter' import {awsClientsForAccount, resolveAccountId} from './aws' import {createEc2Service} from './services/ec2' @@ -103,6 +104,7 @@ export function createCloudAdapterRegistry(accountId?: string | null): CloudAdap new AwsUserPoolsAdapter(clients.cognito), new AwsDnsAdapter(clients.route53), new AwsCdnAdapter(clients.cloudfront, clients.s3), + new AwsPipesAdapter(clients.pipes), new AwsMonitoringAdapter(clients.cloudwatch), new AzureStorageAdapter(), new AzureServiceBusAdapter(), diff --git a/packages/frontend/src/components/DynamicResourceView.tsx b/packages/frontend/src/components/DynamicResourceView.tsx index e97677d..3bc7fd7 100644 --- a/packages/frontend/src/components/DynamicResourceView.tsx +++ b/packages/frontend/src/components/DynamicResourceView.tsx @@ -68,6 +68,7 @@ import { LoadBalancerListenerPanel } from "@/components/LoadBalancerListenerPane import { CognitoUsersPanel } from "@/components/CognitoUsersPanel"; import { Route53RecordsPanel } from "@/components/Route53RecordsPanel"; import { CloudFrontPanel } from "@/components/CloudFrontPanel"; +import { PipePanel } from "@/components/PipePanel"; import { CloudWatchAlarmPanel } from "@/components/CloudWatchAlarmPanel"; import { CloudWatchMetricsExplorer } from "@/components/CloudWatchMetricsExplorer"; import { WorkflowExecutionDebugger } from "@/components/WorkflowExecutionDebugger"; @@ -987,6 +988,9 @@ export function DynamicResourceView({ {service === "dns" && cloud === "aws" && ( )} + {service === "pipes" && cloud === "aws" && ( + + )} {service === "cdn" && cloud === "aws" && ( )} diff --git a/packages/frontend/src/components/PipePanel.test.tsx b/packages/frontend/src/components/PipePanel.test.tsx new file mode 100644 index 0000000..db7e622 --- /dev/null +++ b/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 = {}): 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() +} + +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() +}) diff --git a/packages/frontend/src/components/PipePanel.tsx b/packages/frontend/src/components/PipePanel.tsx new file mode 100644 index 0000000..817f71d --- /dev/null +++ b/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 = { + 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) }}> +