feat: EC2 UI polished + storage persisting
This commit is contained in:
@@ -112,12 +112,16 @@ describe('AwsComputeAdapter', () => {
|
||||
keyName: 'my-key',
|
||||
subnetId: 'subnet-abc',
|
||||
securityGroupIds: 'sg-111, sg-222',
|
||||
rootVolumeSize: '30',
|
||||
rootVolumeType: 'gp3',
|
||||
}})
|
||||
expect(launched).toHaveLength(1)
|
||||
expect(launched[0].name).toBe('test-box')
|
||||
expect(launched[0].imageId).toBe('ami-0abcdef123')
|
||||
expect(launched[0].instanceType).toBe('t3.micro')
|
||||
expect(launched[0].securityGroupIds).toEqual(['sg-111', 'sg-222'])
|
||||
expect(launched[0].rootVolumeSize).toBe(30)
|
||||
expect(launched[0].rootVolumeType).toBe('gp3')
|
||||
expect(result.type).toBe('instance')
|
||||
expect(result.cloud).toBe('aws')
|
||||
})
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import {awsComputeSchema} from '../cloud-spi/computeSchema'
|
||||
import {ValidationError} from '../cloud-spi/errors'
|
||||
import type {
|
||||
CloudResource,
|
||||
CloudServiceAdapter,
|
||||
@@ -51,6 +52,7 @@ export class AwsComputeAdapter implements CloudServiceAdapter {
|
||||
const v = input.values
|
||||
const sgRaw = typeof v.securityGroupIds === 'string' ? v.securityGroupIds : ''
|
||||
const securityGroupIds = sgRaw ? sgRaw.split(',').map((s: string) => s.trim()).filter(Boolean) : undefined
|
||||
const rootVolumeSize = positiveInteger(v.rootVolumeSize)
|
||||
const runInput: RunInstanceInput = {
|
||||
name: String(v.name ?? ''),
|
||||
imageId: String(v.imageId ?? ''),
|
||||
@@ -58,6 +60,8 @@ export class AwsComputeAdapter implements CloudServiceAdapter {
|
||||
keyName: v.keyName ? String(v.keyName) : undefined,
|
||||
subnetId: v.subnetId ? String(v.subnetId) : undefined,
|
||||
securityGroupIds,
|
||||
rootVolumeSize,
|
||||
rootVolumeType: v.rootVolumeType ? String(v.rootVolumeType) : undefined,
|
||||
}
|
||||
const instance = await this.service_.runInstance(runInput)
|
||||
return instanceToResource(instance)
|
||||
@@ -68,6 +72,13 @@ export class AwsComputeAdapter implements CloudServiceAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
function positiveInteger(value: unknown): number | undefined {
|
||||
if (value === undefined || value === null || String(value).trim() === '') return undefined
|
||||
const parsed = Number(value)
|
||||
if (!Number.isInteger(parsed) || parsed <= 0) throw new ValidationError('rootVolumeSize must be a positive whole number')
|
||||
return parsed
|
||||
}
|
||||
|
||||
function instanceToResource(instance: Ec2Instance): CloudResource {
|
||||
return {
|
||||
id: instance.instanceId,
|
||||
|
||||
@@ -2,10 +2,17 @@ import {describe, expect, test} from 'bun:test'
|
||||
import {
|
||||
CreateEventBusCommand,
|
||||
DeleteEventBusCommand,
|
||||
DeleteRuleCommand,
|
||||
DescribeEventBusCommand,
|
||||
ListEventBusesCommand,
|
||||
ListRulesCommand,
|
||||
ListTargetsByRuleCommand,
|
||||
PutRuleCommand,
|
||||
PutTargetsCommand,
|
||||
RemoveTargetsCommand,
|
||||
type EventBridgeClient,
|
||||
} from '@aws-sdk/client-eventbridge'
|
||||
import {AddPermissionCommand, type LambdaClient} from '@aws-sdk/client-lambda'
|
||||
import {AwsEventBridgeAdapter} from './AwsEventBridgeAdapter'
|
||||
|
||||
type SendHandler = (command: unknown) => Promise<unknown>
|
||||
@@ -74,8 +81,64 @@ describe('AwsEventBridgeAdapter', () => {
|
||||
expect(commands[1]).toBeInstanceOf(DeleteEventBusCommand)
|
||||
expect((commands[1] as DeleteEventBusCommand).input.Name).toBe('orders')
|
||||
})
|
||||
|
||||
test('lists every rule and its targets for an event bus', async () => {
|
||||
const adapter = createAdapter(async (command) => {
|
||||
if (command instanceof ListRulesCommand) {
|
||||
expect(command.input.EventBusName).toBe('orders')
|
||||
return command.input.NextToken
|
||||
? {Rules: [{Name: 'nightly', Arn: 'arn:aws:events:us-east-1:000000000000:rule/orders/nightly', EventPattern: '{"source":["billing"]}', State: 'DISABLED'}]}
|
||||
: {Rules: [{Name: 'hourly', Arn: 'arn:aws:events:us-east-1:000000000000:rule/orders/hourly', ScheduleExpression: 'rate(1 hour)'}], NextToken: 'next'}
|
||||
}
|
||||
if (command instanceof ListTargetsByRuleCommand) {
|
||||
return {Targets: [{Id: 'worker', Arn: 'arn:aws:lambda:us-east-1:000000000000:function:worker'}]}
|
||||
}
|
||||
return {}
|
||||
})
|
||||
|
||||
await expect(adapter.listEventBridgeRules('event-bus:orders')).resolves.toEqual([
|
||||
{name: 'hourly', busName: 'orders', busId: 'event-bus:orders', arn: 'arn:aws:events:us-east-1:000000000000:rule/orders/hourly', scheduleExpression: 'rate(1 hour)', state: 'ENABLED', targets: [{id: 'worker', arn: 'arn:aws:lambda:us-east-1:000000000000:function:worker'}]},
|
||||
{name: 'nightly', busName: 'orders', busId: 'event-bus:orders', arn: 'arn:aws:events:us-east-1:000000000000:rule/orders/nightly', eventPattern: '{"source":["billing"]}', state: 'DISABLED', targets: [{id: 'worker', arn: 'arn:aws:lambda:us-east-1:000000000000:function:worker'}]},
|
||||
])
|
||||
})
|
||||
|
||||
test('creates a rule, attaches targets, and permits Lambda invocation', async () => {
|
||||
const commands: unknown[] = []
|
||||
const adapter = createAdapter(async (command) => {
|
||||
commands.push(command)
|
||||
if (command instanceof PutRuleCommand) return {RuleArn: 'arn:aws:events:us-east-1:000000000000:rule/orders/hourly'}
|
||||
return {}
|
||||
})
|
||||
|
||||
await expect(adapter.createEventBridgeRule('event-bus:orders', {
|
||||
name: 'hourly', scheduleExpression: 'rate(1 hour)', state: 'DISABLED', description: 'Run the worker',
|
||||
targets: [{id: 'worker', arn: 'arn:aws:lambda:us-east-1:000000000000:function:worker'}],
|
||||
})).resolves.toMatchObject({name: 'hourly', busName: 'orders', state: 'DISABLED'})
|
||||
|
||||
expect((commands[0] as PutRuleCommand).input).toEqual({Name: 'hourly', EventBusName: 'orders', ScheduleExpression: 'rate(1 hour)', State: 'DISABLED', Description: 'Run the worker'})
|
||||
expect((commands[1] as PutTargetsCommand).input).toEqual({Rule: 'hourly', EventBusName: 'orders', Targets: [{Id: 'worker', Arn: 'arn:aws:lambda:us-east-1:000000000000:function:worker'}]})
|
||||
expect((commands[2] as AddPermissionCommand).input).toMatchObject({FunctionName: 'arn:aws:lambda:us-east-1:000000000000:function:worker', Principal: 'events.amazonaws.com', SourceArn: 'arn:aws:events:us-east-1:000000000000:rule/orders/hourly'})
|
||||
})
|
||||
|
||||
test('removes targets before deleting a rule', async () => {
|
||||
const commands: unknown[] = []
|
||||
const adapter = createAdapter(async (command) => {
|
||||
commands.push(command)
|
||||
if (command instanceof ListTargetsByRuleCommand) return {Targets: [{Id: 'worker', Arn: 'arn:aws:lambda:us-east-1:000000000000:function:worker'}]}
|
||||
return {}
|
||||
})
|
||||
|
||||
await adapter.deleteEventBridgeRule('event-bus:orders', 'hourly')
|
||||
|
||||
expect(commands[0]).toBeInstanceOf(ListTargetsByRuleCommand)
|
||||
expect((commands[1] as RemoveTargetsCommand).input).toEqual({Rule: 'hourly', EventBusName: 'orders', Ids: ['worker']})
|
||||
expect((commands[2] as DeleteRuleCommand).input).toEqual({Name: 'hourly', EventBusName: 'orders'})
|
||||
})
|
||||
})
|
||||
|
||||
function createAdapter(handler: SendHandler = async () => ({EventBuses: []})): AwsEventBridgeAdapter {
|
||||
return new AwsEventBridgeAdapter({send: handler} as unknown as EventBridgeClient)
|
||||
return new AwsEventBridgeAdapter(
|
||||
{send: handler} as unknown as EventBridgeClient,
|
||||
{send: handler} as unknown as LambdaClient,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -2,10 +2,19 @@ import {
|
||||
CreateEventBusCommand,
|
||||
DeleteEventBusCommand,
|
||||
DescribeEventBusCommand,
|
||||
DeleteRuleCommand,
|
||||
ListEventBusesCommand,
|
||||
ListRulesCommand,
|
||||
ListTargetsByRuleCommand,
|
||||
PutRuleCommand,
|
||||
PutTargetsCommand,
|
||||
RemoveTargetsCommand,
|
||||
type EventBridgeClient,
|
||||
type EventBus,
|
||||
type Rule,
|
||||
type Target,
|
||||
} from '@aws-sdk/client-eventbridge'
|
||||
import {AddPermissionCommand, type LambdaClient} from '@aws-sdk/client-lambda'
|
||||
import {awsEventBridgeSchema} from '../cloud-spi/eventbridgeSchema'
|
||||
import {ValidationError} from '../cloud-spi/errors'
|
||||
import type {
|
||||
@@ -13,6 +22,9 @@ import type {
|
||||
CloudServiceAdapter,
|
||||
CloudServiceDescriptorOverride,
|
||||
CreateResourceInput,
|
||||
EventBridgeRuleInput,
|
||||
EventBridgeRuleOverview,
|
||||
EventBridgeRuleTarget,
|
||||
ResourceQuery,
|
||||
ServiceSchema,
|
||||
} from '../cloud-spi/types'
|
||||
@@ -24,7 +36,10 @@ export class AwsEventBridgeAdapter implements CloudServiceAdapter {
|
||||
readonly cloud = 'aws' as const
|
||||
readonly service = 'events' as const
|
||||
|
||||
constructor(private readonly eventbridge: EventBridgeClient) {}
|
||||
constructor(
|
||||
private readonly eventbridge: EventBridgeClient,
|
||||
private readonly lambda: LambdaClient,
|
||||
) {}
|
||||
|
||||
schema(): ServiceSchema {
|
||||
return awsEventBridgeSchema()
|
||||
@@ -67,6 +82,90 @@ export class AwsEventBridgeAdapter implements CloudServiceAdapter {
|
||||
async delete(id: string): Promise<void> {
|
||||
await this.eventbridge.send(new DeleteEventBusCommand({Name: eventBusName(id)}))
|
||||
}
|
||||
|
||||
async listEventBridgeRules(eventBusId: string): Promise<EventBridgeRuleOverview[]> {
|
||||
const busName = eventBusName(eventBusId)
|
||||
const rules: Rule[] = []
|
||||
let nextToken: string | undefined
|
||||
do {
|
||||
const response = await this.eventbridge.send(new ListRulesCommand({
|
||||
...eventBusInput(busName),
|
||||
NextToken: nextToken,
|
||||
}))
|
||||
rules.push(...(response.Rules ?? []))
|
||||
nextToken = response.NextToken
|
||||
} while (nextToken)
|
||||
|
||||
return Promise.all(rules.filter((rule) => Boolean(rule.Name)).map(async (rule) => {
|
||||
const name = rule.Name as string
|
||||
return ruleOverview(rule, busName, await this.listTargets(name, busName))
|
||||
}))
|
||||
}
|
||||
|
||||
async createEventBridgeRule(eventBusId: string, input: EventBridgeRuleInput): Promise<EventBridgeRuleOverview> {
|
||||
const busName = eventBusName(eventBusId)
|
||||
const name = requiredValue(input.name, 'name')
|
||||
const targetInput = ruleInput(input)
|
||||
const response = await this.eventbridge.send(new PutRuleCommand({
|
||||
Name: name,
|
||||
...eventBusInput(busName),
|
||||
...(targetInput.scheduleExpression ? {ScheduleExpression: targetInput.scheduleExpression} : {}),
|
||||
...(targetInput.eventPattern ? {EventPattern: targetInput.eventPattern} : {}),
|
||||
State: targetInput.state,
|
||||
...(targetInput.description ? {Description: targetInput.description} : {}),
|
||||
}))
|
||||
const arn = response.RuleArn
|
||||
if (!arn) throw new ValidationError(`EventBridge rule ${name} did not return an ARN`)
|
||||
|
||||
await this.eventbridge.send(new PutTargetsCommand({
|
||||
Rule: name,
|
||||
...eventBusInput(busName),
|
||||
Targets: input.targets.map((target) => ({Id: target.id, Arn: target.arn})),
|
||||
}))
|
||||
await Promise.all(input.targets.filter((target) => isLambdaArn(target.arn)).map((target) =>
|
||||
this.lambda.send(new AddPermissionCommand({
|
||||
FunctionName: target.arn,
|
||||
StatementId: permissionStatementId(busName, name, target.id),
|
||||
Action: 'lambda:InvokeFunction',
|
||||
Principal: 'events.amazonaws.com',
|
||||
SourceArn: arn,
|
||||
})),
|
||||
))
|
||||
return {
|
||||
name,
|
||||
busName,
|
||||
busId: `${EVENT_BUS_PREFIX}${busName}`,
|
||||
arn,
|
||||
...targetInput,
|
||||
targets: input.targets,
|
||||
}
|
||||
}
|
||||
|
||||
async deleteEventBridgeRule(eventBusId: string, ruleName: string): Promise<void> {
|
||||
const busName = eventBusName(eventBusId)
|
||||
const name = requiredValue(ruleName, 'ruleName')
|
||||
const targets = await this.listTargets(name, busName)
|
||||
const targetIds = targets.map((target) => target.Id).filter((id): id is string => Boolean(id))
|
||||
if (targetIds.length > 0) {
|
||||
await this.eventbridge.send(new RemoveTargetsCommand({Rule: name, ...eventBusInput(busName), Ids: targetIds}))
|
||||
}
|
||||
await this.eventbridge.send(new DeleteRuleCommand({Name: name, ...eventBusInput(busName)}))
|
||||
}
|
||||
|
||||
private async listTargets(ruleName: string, busName: string): Promise<Target[]> {
|
||||
const targets: Target[] = []
|
||||
let nextToken: string | undefined
|
||||
do {
|
||||
const response = await this.eventbridge.send(new ListTargetsByRuleCommand({
|
||||
Rule: ruleName,
|
||||
...eventBusInput(busName),
|
||||
NextToken: nextToken,
|
||||
}))
|
||||
targets.push(...(response.Targets ?? []))
|
||||
nextToken = response.NextToken
|
||||
} while (nextToken)
|
||||
return targets
|
||||
}
|
||||
}
|
||||
|
||||
function eventBusResource(eventBus: EventBus): CloudResource {
|
||||
@@ -122,3 +221,55 @@ function isNotFound(error: unknown): boolean {
|
||||
const value = error as {name?: string; $metadata?: {httpStatusCode?: number}}
|
||||
return value.name === 'ResourceNotFoundException' || value.$metadata?.httpStatusCode === 404
|
||||
}
|
||||
|
||||
function eventBusInput(busName: string): {EventBusName?: string} {
|
||||
return busName === 'default' ? {} : {EventBusName: busName}
|
||||
}
|
||||
|
||||
function ruleInput(input: EventBridgeRuleInput): Pick<EventBridgeRuleOverview, 'scheduleExpression' | 'eventPattern' | 'state' | 'description'> {
|
||||
const scheduleExpression = stringValue(input.scheduleExpression)
|
||||
const eventPattern = stringValue(input.eventPattern)
|
||||
if (Boolean(scheduleExpression) === Boolean(eventPattern)) {
|
||||
throw new ValidationError('exactly one of scheduleExpression or eventPattern is required')
|
||||
}
|
||||
if (eventPattern) {
|
||||
try {
|
||||
JSON.parse(eventPattern)
|
||||
} catch {
|
||||
throw new ValidationError('eventPattern must be a JSON string')
|
||||
}
|
||||
}
|
||||
if (!Array.isArray(input.targets) || input.targets.length === 0) throw new ValidationError('at least one target is required')
|
||||
const targetsValid = input.targets.every((target) => requiredValue(target.id, 'target.id') && requiredValue(target.arn, 'target.arn'))
|
||||
if (!targetsValid) throw new ValidationError('target id and arn are required')
|
||||
const state = input.state ?? 'ENABLED'
|
||||
if (state !== 'ENABLED' && state !== 'DISABLED') throw new ValidationError('state must be ENABLED or DISABLED')
|
||||
return {
|
||||
...(scheduleExpression ? {scheduleExpression} : {}),
|
||||
...(eventPattern ? {eventPattern} : {}),
|
||||
state,
|
||||
...(stringValue(input.description) ? {description: stringValue(input.description)} : {}),
|
||||
}
|
||||
}
|
||||
|
||||
function ruleOverview(rule: Rule, busName: string, targets: Target[]): EventBridgeRuleOverview {
|
||||
return {
|
||||
name: rule.Name ?? '',
|
||||
busName,
|
||||
busId: `${EVENT_BUS_PREFIX}${busName}`,
|
||||
arn: rule.Arn ?? null,
|
||||
...(rule.ScheduleExpression ? {scheduleExpression: rule.ScheduleExpression} : {}),
|
||||
...(rule.EventPattern ? {eventPattern: rule.EventPattern} : {}),
|
||||
state: rule.State === 'DISABLED' ? 'DISABLED' : 'ENABLED',
|
||||
...(rule.Description ? {description: rule.Description} : {}),
|
||||
targets: targets.flatMap((target) => target.Id && target.Arn ? [{id: target.Id, arn: target.Arn}] : []),
|
||||
}
|
||||
}
|
||||
|
||||
function isLambdaArn(arn: string): boolean {
|
||||
return arn.split(':')[2] === 'lambda'
|
||||
}
|
||||
|
||||
function permissionStatementId(busName: string, ruleName: string, targetId: string): string {
|
||||
return `floci-events-${busName}-${ruleName}-${targetId}`.replace(/[^A-Za-z0-9_-]/g, '_').slice(0, 100)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
import {describe, expect, test} from 'bun:test'
|
||||
import {
|
||||
CreateTopicCommand,
|
||||
DeleteTopicCommand,
|
||||
GetSubscriptionAttributesCommand,
|
||||
GetTopicAttributesCommand,
|
||||
ListSubscriptionsByTopicCommand,
|
||||
ListTopicsCommand,
|
||||
PublishCommand,
|
||||
SubscribeCommand,
|
||||
type SNSClient,
|
||||
UnsubscribeCommand,
|
||||
} from '@aws-sdk/client-sns'
|
||||
import {AwsSnsAdapter} from './AwsSnsAdapter'
|
||||
import {ValidationError} from '../cloud-spi/errors'
|
||||
|
||||
const ORDERS_ARN = 'arn:aws:sns:us-east-1:000000000000:orders'
|
||||
const ALERTS_ARN = 'arn:aws:sns:us-east-1:000000000000:alerts'
|
||||
|
||||
function stubSns() {
|
||||
const sent: object[] = []
|
||||
const client = {
|
||||
async send(command: object) {
|
||||
sent.push(command)
|
||||
if (command instanceof ListTopicsCommand) return {Topics: [{TopicArn: ORDERS_ARN}, {TopicArn: ALERTS_ARN}]}
|
||||
if (command instanceof GetTopicAttributesCommand) {
|
||||
return {Attributes: {DisplayName: 'Order notifications', FifoTopic: 'false'}}
|
||||
}
|
||||
if (command instanceof CreateTopicCommand) {
|
||||
return {TopicArn: `arn:aws:sns:us-east-1:000000000000:${command.input.Name}`}
|
||||
}
|
||||
if (command instanceof PublishCommand) return {MessageId: 'e5a98c5b-cd94-4c38-a9ac-7260ec3e575f'}
|
||||
if (command instanceof ListSubscriptionsByTopicCommand) return {Subscriptions: [{
|
||||
SubscriptionArn: 'arn:aws:sns:us-east-1:000000000000:orders:sub-1',
|
||||
Protocol: 'sqs',
|
||||
Endpoint: 'arn:aws:sqs:us-east-1:000000000000:orders',
|
||||
}]}
|
||||
if (command instanceof GetSubscriptionAttributesCommand) return {Attributes: {FilterPolicy: '{"eventType":["created"]}'}}
|
||||
if (command instanceof SubscribeCommand) return {SubscriptionArn: 'arn:aws:sns:us-east-1:000000000000:orders:sub-2'}
|
||||
return {}
|
||||
},
|
||||
} as unknown as SNSClient
|
||||
return {client, sent}
|
||||
}
|
||||
|
||||
describe('AwsSnsAdapter', () => {
|
||||
test('lists SNS topic ARNs as topic resources', async () => {
|
||||
const {client} = stubSns()
|
||||
const [topic] = await new AwsSnsAdapter(client).list()
|
||||
|
||||
expect(topic).toMatchObject({id: ORDERS_ARN, name: 'orders', service: 'notifications', type: 'topic'})
|
||||
expect(topic?.metadata.arn).toBe(ORDERS_ARN)
|
||||
})
|
||||
|
||||
test('gets a topic with its SNS attributes', async () => {
|
||||
const {client, sent} = stubSns()
|
||||
const topic = await new AwsSnsAdapter(client).get(ORDERS_ARN)
|
||||
|
||||
expect(sent[0]).toBeInstanceOf(GetTopicAttributesCommand)
|
||||
expect(topic?.metadata.displayName).toBe('Order notifications')
|
||||
})
|
||||
|
||||
test('creates and reads back an SNS topic', async () => {
|
||||
const {client, sent} = stubSns()
|
||||
const topic = await new AwsSnsAdapter(client).create({values: {name: 'orders'}})
|
||||
|
||||
expect(sent[0]).toBeInstanceOf(CreateTopicCommand)
|
||||
expect((sent[0] as CreateTopicCommand).input.Name).toBe('orders')
|
||||
expect(sent[1]).toBeInstanceOf(GetTopicAttributesCommand)
|
||||
expect(topic.id).toBe(ORDERS_ARN)
|
||||
})
|
||||
|
||||
test('deletes a topic by ARN', async () => {
|
||||
const {client, sent} = stubSns()
|
||||
await new AwsSnsAdapter(client).delete(ORDERS_ARN)
|
||||
|
||||
expect((sent[0] as DeleteTopicCommand).input.TopicArn).toBe(ORDERS_ARN)
|
||||
})
|
||||
|
||||
test('publishes a message with optional subject and string message attributes', async () => {
|
||||
const {client, sent} = stubSns()
|
||||
const result = await new AwsSnsAdapter(client).publish(ORDERS_ARN, {
|
||||
message: 'Order placed',
|
||||
subject: 'Order update',
|
||||
messageAttributes: {orderId: '1234'},
|
||||
})
|
||||
|
||||
expect(result.messageId).toBe('e5a98c5b-cd94-4c38-a9ac-7260ec3e575f')
|
||||
expect((sent[0] as PublishCommand).input).toEqual({
|
||||
TopicArn: ORDERS_ARN,
|
||||
Message: 'Order placed',
|
||||
Subject: 'Order update',
|
||||
MessageAttributes: {orderId: {DataType: 'String', StringValue: '1234'}},
|
||||
})
|
||||
})
|
||||
|
||||
test('manages SNS subscriptions with direct SDK commands and exposes their attributes', async () => {
|
||||
const {client, sent} = stubSns()
|
||||
const adapter = new AwsSnsAdapter(client)
|
||||
|
||||
const created = await adapter.createSnsSubscription(ORDERS_ARN, {
|
||||
protocol: 'sqs',
|
||||
endpoint: 'arn:aws:sqs:us-east-1:000000000000:orders',
|
||||
})
|
||||
const subscriptions = await adapter.listSnsSubscriptions(ORDERS_ARN)
|
||||
await adapter.deleteSnsSubscription('arn:aws:sns:us-east-1:000000000000:orders:sub-1')
|
||||
|
||||
expect((sent[0] as SubscribeCommand).input).toEqual({
|
||||
TopicArn: ORDERS_ARN,
|
||||
Protocol: 'sqs',
|
||||
Endpoint: 'arn:aws:sqs:us-east-1:000000000000:orders',
|
||||
})
|
||||
expect((sent[1] as ListSubscriptionsByTopicCommand).input).toEqual({TopicArn: ORDERS_ARN})
|
||||
expect((sent[2] as GetSubscriptionAttributesCommand).input).toEqual({
|
||||
SubscriptionArn: 'arn:aws:sns:us-east-1:000000000000:orders:sub-1',
|
||||
})
|
||||
expect((sent[3] as UnsubscribeCommand).input).toEqual({
|
||||
SubscriptionArn: 'arn:aws:sns:us-east-1:000000000000:orders:sub-1',
|
||||
})
|
||||
expect(created).toMatchObject({arn: 'arn:aws:sns:us-east-1:000000000000:orders:sub-2', protocol: 'sqs', status: 'Confirmed'})
|
||||
expect(subscriptions).toEqual([{
|
||||
arn: 'arn:aws:sns:us-east-1:000000000000:orders:sub-1',
|
||||
protocol: 'sqs',
|
||||
endpoint: 'arn:aws:sqs:us-east-1:000000000000:orders',
|
||||
status: 'Confirmed',
|
||||
filterPolicy: '{"eventType":["created"]}',
|
||||
}])
|
||||
})
|
||||
|
||||
test('rejects invalid topic names and blank messages before calling SNS', async () => {
|
||||
const {client, sent} = stubSns()
|
||||
const adapter = new AwsSnsAdapter(client)
|
||||
|
||||
await expect(adapter.create({values: {name: 'has space'}})).rejects.toBeInstanceOf(ValidationError)
|
||||
await expect(adapter.publish(ORDERS_ARN, {message: ''})).rejects.toBeInstanceOf(ValidationError)
|
||||
expect(sent).toEqual([])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,200 @@
|
||||
import {
|
||||
CreateTopicCommand,
|
||||
DeleteTopicCommand,
|
||||
GetSubscriptionAttributesCommand,
|
||||
GetTopicAttributesCommand,
|
||||
ListSubscriptionsByTopicCommand,
|
||||
ListTopicsCommand,
|
||||
PublishCommand,
|
||||
SubscribeCommand,
|
||||
type SNSClient,
|
||||
type Subscription,
|
||||
type Topic,
|
||||
UnsubscribeCommand,
|
||||
} from '@aws-sdk/client-sns'
|
||||
import {ValidationError} from '../cloud-spi/errors'
|
||||
import {awsNotificationsSchema} from '../cloud-spi/notificationsSchema'
|
||||
import type {
|
||||
CloudResource,
|
||||
CloudServiceAdapter,
|
||||
CreateResourceInput,
|
||||
PublishNotificationInput,
|
||||
PublishNotificationResult,
|
||||
ResourceQuery,
|
||||
ServiceSchema,
|
||||
SnsSubscriptionInput,
|
||||
SnsSubscriptionOverview,
|
||||
} from '../cloud-spi/types'
|
||||
|
||||
/** Amazon SNS topics through the provider-neutral Notifications category. */
|
||||
export class AwsSnsAdapter implements CloudServiceAdapter {
|
||||
readonly cloud = 'aws' as const
|
||||
readonly service = 'notifications' as const
|
||||
|
||||
constructor(private readonly sns: SNSClient) {}
|
||||
|
||||
schema(): ServiceSchema {
|
||||
return awsNotificationsSchema()
|
||||
}
|
||||
|
||||
async list(query: ResourceQuery = {}): Promise<CloudResource[]> {
|
||||
const topics: Topic[] = []
|
||||
let nextToken: string | undefined
|
||||
do {
|
||||
const response = await this.sns.send(new ListTopicsCommand(nextToken ? {NextToken: nextToken} : {}))
|
||||
topics.push(...(response.Topics ?? []))
|
||||
nextToken = response.NextToken
|
||||
} while (nextToken)
|
||||
|
||||
return filterBySearch(topics.map((topic) => topicResource(topic)), query.search)
|
||||
}
|
||||
|
||||
async get(id: string): Promise<CloudResource | null> {
|
||||
try {
|
||||
const response = await this.sns.send(new GetTopicAttributesCommand({TopicArn: id}))
|
||||
return topicResource({TopicArn: id}, response.Attributes)
|
||||
} catch (error) {
|
||||
if (isMissing(error)) return null
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async create(input: CreateResourceInput): Promise<CloudResource> {
|
||||
const name = requiredName(input.values.name)
|
||||
const response = await this.sns.send(new CreateTopicCommand({Name: name}))
|
||||
const arn = response.TopicArn
|
||||
if (!arn) throw new ValidationError(`SNS did not return an ARN for ${name}`)
|
||||
return (await this.get(arn)) ?? topicResource({TopicArn: arn})
|
||||
}
|
||||
|
||||
async delete(id: string): Promise<void> {
|
||||
await this.sns.send(new DeleteTopicCommand({TopicArn: id}))
|
||||
}
|
||||
|
||||
async publish(id: string, input: PublishNotificationInput): Promise<PublishNotificationResult> {
|
||||
const message = requiredMessage(input.message)
|
||||
const subject = optionalSubject(input.subject)
|
||||
const MessageAttributes = toMessageAttributes(input.messageAttributes)
|
||||
const response = await this.sns.send(new PublishCommand({
|
||||
TopicArn: id,
|
||||
Message: message,
|
||||
...(subject ? {Subject: subject} : {}),
|
||||
...(MessageAttributes ? {MessageAttributes} : {}),
|
||||
}))
|
||||
return {messageId: response.MessageId ?? null}
|
||||
}
|
||||
|
||||
async createSnsSubscription(topicArn: string, input: SnsSubscriptionInput): Promise<SnsSubscriptionOverview> {
|
||||
const protocol = requiredSubscriptionValue(input.protocol, 'protocol')
|
||||
const endpoint = requiredSubscriptionValue(input.endpoint, 'endpoint')
|
||||
const response = await this.sns.send(new SubscribeCommand({
|
||||
TopicArn: topicArn,
|
||||
Protocol: protocol,
|
||||
Endpoint: endpoint,
|
||||
}))
|
||||
return {
|
||||
arn: response.SubscriptionArn ?? null,
|
||||
protocol,
|
||||
endpoint,
|
||||
status: response.SubscriptionArn === 'PendingConfirmation' ? 'PendingConfirmation' : 'Confirmed',
|
||||
filterPolicy: null,
|
||||
}
|
||||
}
|
||||
|
||||
async listSnsSubscriptions(topicArn: string): Promise<SnsSubscriptionOverview[]> {
|
||||
const subscriptions: Subscription[] = []
|
||||
let nextToken: string | undefined
|
||||
do {
|
||||
const response = await this.sns.send(new ListSubscriptionsByTopicCommand({
|
||||
TopicArn: topicArn,
|
||||
...(nextToken ? {NextToken: nextToken} : {}),
|
||||
}))
|
||||
subscriptions.push(...(response.Subscriptions ?? []))
|
||||
nextToken = response.NextToken
|
||||
} while (nextToken)
|
||||
|
||||
return Promise.all(subscriptions.map((subscription) => this.subscriptionOverview(subscription)))
|
||||
}
|
||||
|
||||
async deleteSnsSubscription(subscriptionArn: string): Promise<void> {
|
||||
await this.sns.send(new UnsubscribeCommand({SubscriptionArn: subscriptionArn}))
|
||||
}
|
||||
|
||||
async health(): Promise<void> {
|
||||
await this.sns.send(new ListTopicsCommand({}))
|
||||
}
|
||||
|
||||
private async subscriptionOverview(subscription: Subscription): Promise<SnsSubscriptionOverview> {
|
||||
const arn = subscription.SubscriptionArn ?? null
|
||||
const pending = arn === 'PendingConfirmation'
|
||||
const attributes = !arn || pending
|
||||
? undefined
|
||||
: await this.sns.send(new GetSubscriptionAttributesCommand({SubscriptionArn: arn}))
|
||||
return {
|
||||
arn,
|
||||
protocol: subscription.Protocol ?? '',
|
||||
endpoint: subscription.Endpoint ?? '',
|
||||
status: pending ? 'PendingConfirmation' : 'Confirmed',
|
||||
filterPolicy: attributes?.Attributes?.FilterPolicy ?? null,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function topicResource(topic: Topic, attributes: Record<string, string> = {}): CloudResource {
|
||||
const arn = topic.TopicArn ?? ''
|
||||
return {
|
||||
id: arn,
|
||||
name: topicName(arn),
|
||||
cloud: 'aws',
|
||||
service: 'notifications',
|
||||
type: attributes.FifoTopic === 'true' ? 'fifo-topic' : 'topic',
|
||||
region: arn.split(':')[3] || null,
|
||||
createdAt: null,
|
||||
metadata: {arn, fifoTopic: attributes.FifoTopic === 'true', displayName: attributes.DisplayName},
|
||||
}
|
||||
}
|
||||
|
||||
function topicName(arn: string): string {
|
||||
return arn.split(':').pop() ?? arn
|
||||
}
|
||||
|
||||
function filterBySearch(resources: CloudResource[], search?: string): CloudResource[] {
|
||||
const normalized = search?.trim().toLowerCase()
|
||||
if (!normalized) return resources
|
||||
return resources.filter((resource) => resource.name.toLowerCase().includes(normalized))
|
||||
}
|
||||
|
||||
function requiredName(value: unknown): string {
|
||||
if (typeof value !== 'string' || !value.trim()) throw new ValidationError('name is required')
|
||||
const name = value.trim()
|
||||
if (!/^[A-Za-z0-9_-]{1,256}$/.test(name) && !/^[A-Za-z0-9_-]{1,251}\.fifo$/.test(name)) {
|
||||
throw new ValidationError('name must be up to 256 letters, numbers, hyphens, underscores, or end in .fifo')
|
||||
}
|
||||
return name
|
||||
}
|
||||
|
||||
function requiredMessage(value: unknown): string {
|
||||
if (typeof value !== 'string' || !value) throw new ValidationError('message is required')
|
||||
return value
|
||||
}
|
||||
|
||||
function requiredSubscriptionValue(value: unknown, field: string): string {
|
||||
if (typeof value !== 'string' || !value.trim()) throw new ValidationError(`${field} is required`)
|
||||
return value.trim()
|
||||
}
|
||||
|
||||
function optionalSubject(value: unknown): string | undefined {
|
||||
if (value === undefined || value === '') return undefined
|
||||
if (typeof value !== 'string') throw new ValidationError('subject must be a string')
|
||||
if (value.length > 100) throw new ValidationError('subject must be 100 characters or fewer')
|
||||
return value
|
||||
}
|
||||
|
||||
function toMessageAttributes(values: Record<string, string> | undefined) {
|
||||
if (!values || Object.keys(values).length === 0) return undefined
|
||||
return Object.fromEntries(Object.entries(values).map(([Name, Value]) => [Name, {DataType: 'String', StringValue: Value}]))
|
||||
}
|
||||
|
||||
function isMissing(error: unknown): boolean {
|
||||
return typeof error === 'object' && error !== null && (error as {name?: string}).name === 'NotFound'
|
||||
}
|
||||
Reference in New Issue
Block a user