diff --git a/packages/api/package.json b/packages/api/package.json index 4c60859..9ce7d92 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -16,6 +16,7 @@ "@aws-sdk/client-rds": "^3.1076.0", "@aws-sdk/client-s3": "^3.1076.0", "@aws-sdk/client-secrets-manager": "^3.1076.0", + "@aws-sdk/client-sqs": "^3.1090.0", "dotenv": "^17.4.2", "hono": "^4.12.27" }, diff --git a/packages/api/src/adapter-aws/AwsQueueAdapter.test.ts b/packages/api/src/adapter-aws/AwsQueueAdapter.test.ts new file mode 100644 index 0000000..f4f5ffd --- /dev/null +++ b/packages/api/src/adapter-aws/AwsQueueAdapter.test.ts @@ -0,0 +1,285 @@ +import {describe, expect, test, vi} from 'bun:test' +import { + CreateQueueCommand, + DeleteMessageCommand, + DeleteQueueCommand, + GetQueueAttributesCommand, + ListQueuesCommand, + PurgeQueueCommand, + ReceiveMessageCommand, + SendMessageCommand, + type SQSClient, +} from '@aws-sdk/client-sqs' +import {AwsQueueAdapter} from './AwsQueueAdapter' + +function mockSqs(handler: (command: unknown) => unknown): SQSClient { + return {send: vi.fn(handler)} as unknown as SQSClient +} + +const QUEUE_URL = 'http://localhost:4566/000000000000/orders-queue' + +describe('AwsQueueAdapter', () => { + test('lists queues and normalizes them to queue resources', async () => { + const sqs = mockSqs((command) => { + if (command instanceof ListQueuesCommand) { + return {QueueUrls: [QUEUE_URL, 'http://localhost:4566/000000000000/notifications-queue']} + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const resources = await adapter.list() + + expect(resources).toHaveLength(2) + expect(resources[0]).toMatchObject({ + id: 'orders-queue', + name: 'orders-queue', + cloud: 'aws', + service: 'queue', + type: 'queue', + }) + expect(resources[0].metadata).toMatchObject({queueService: 'sqs'}) + }) + + test('filters listed queues by search term', async () => { + const sqs = mockSqs((command) => { + if (command instanceof ListQueuesCommand) { + return {QueueUrls: [QUEUE_URL, 'http://localhost:4566/000000000000/notifications-queue']} + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const resources = await adapter.list({search: 'order'}) + + expect(resources).toHaveLength(1) + expect(resources[0].name).toBe('orders-queue') + }) + + test('creates a queue and returns the resource with the queue url', async () => { + const sqs = mockSqs((command) => { + if (command instanceof CreateQueueCommand) { + return {QueueUrl: QUEUE_URL} + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const resource = await adapter.create({values: {queueName: 'orders-queue'}}) + + expect(resource).toMatchObject({id: 'orders-queue', name: 'orders-queue', type: 'queue'}) + }) + + test('appends .fifo and sets FifoQueue when the FIFO option is selected', async () => { + let captured: unknown + const sqs = mockSqs((command) => { + if (command instanceof CreateQueueCommand) { + captured = command + return {QueueUrl: 'http://localhost:4566/000000000000/orders.fifo'} + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const resource = await adapter.create({values: {queueName: 'orders', fifoQueue: 'true'}}) + + const command = captured as CreateQueueCommand + expect(command.input.QueueName).toBe('orders.fifo') + expect(command.input.Attributes?.FifoQueue).toBe('true') + expect(resource.name).toBe('orders.fifo') + }) + + test('accepts an explicit .fifo queue name without double-suffixing', async () => { + let captured: unknown + const sqs = mockSqs((command) => { + if (command instanceof CreateQueueCommand) { + captured = command + return {QueueUrl: 'http://localhost:4566/000000000000/orders.fifo'} + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + await adapter.create({values: {queueName: 'orders.fifo', fifoQueue: 'true'}}) + + const command = captured as CreateQueueCommand + expect(command.input.QueueName).toBe('orders.fifo') + }) + + test('rejects create without a queue name', async () => { + const sqs = mockSqs(() => ({})) + const adapter = new AwsQueueAdapter(sqs) + + await expect(adapter.create({values: {}})).rejects.toThrow('queueName is required') + }) + + test('rejects invalid queue names', async () => { + const sqs = mockSqs(() => ({})) + const adapter = new AwsQueueAdapter(sqs) + + await expect(adapter.create({values: {queueName: 'bad name!'}})).rejects.toThrow('Use a valid SQS queue name') + }) + + test('deletes a queue by url', async () => { + let captured: unknown + const send = vi.fn(async (command: unknown) => { + captured = command + return {} + }) + const sqs = {send} as unknown as SQSClient + const adapter = new AwsQueueAdapter(sqs) + + await adapter.delete(QUEUE_URL) + + expect(send).toHaveBeenCalledWith(expect.any(DeleteQueueCommand)) + const command = captured as DeleteQueueCommand + expect(command.input.QueueUrl).toBe(QUEUE_URL) + }) + + test('get returns enriched attributes when available', async () => { + const sqs = mockSqs((command) => { + if (command instanceof GetQueueAttributesCommand) { + return { + Attributes: { + CreatedTimestamp: '1700000000', + ApproximateNumberOfMessages: '5', + }, + } + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const resource = await adapter.get(QUEUE_URL) + + expect(resource?.createdAt).toBe(new Date(1700000000 * 1000).toISOString()) + expect(resource?.metadata.approximateNumberOfMessages).toBe('5') + }) + + test('get falls back to a basic resource when attributes are unavailable', async () => { + const sqs = mockSqs((command) => { + if (command instanceof GetQueueAttributesCommand) throw new Error('NotFound') + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const resource = await adapter.get(QUEUE_URL) + + expect(resource).toMatchObject({id: 'orders-queue', name: 'orders-queue', type: 'queue'}) + }) + + test('receives messages and normalizes them', async () => { + let captured: unknown + const sqs = mockSqs((command) => { + if (command instanceof ReceiveMessageCommand) { + captured = command + return { + Messages: [ + { + MessageId: 'msg-1', + Body: 'hello', + ReceiptHandle: 'rh-1', + MD5OfBody: 'abc', + Attributes: {SentTimestamp: '1700000000000'}, + }, + ], + } + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const messages = await adapter.receiveMessages(QUEUE_URL, 10) + + expect(messages).toHaveLength(1) + expect(messages[0]).toMatchObject({ + messageId: 'msg-1', + body: 'hello', + receiptHandle: 'rh-1', + md5OfBody: 'abc', + }) + expect(messages[0].attributes).toMatchObject({SentTimestamp: '1700000000000'}) + + const command = captured as ReceiveMessageCommand + expect(command.input.VisibilityTimeout).toBe(0) + }) + + test('returns an empty list when the queue has no messages', async () => { + const sqs = mockSqs((command) => { + if (command instanceof ReceiveMessageCommand) return {} + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + const messages = await adapter.receiveMessages(QUEUE_URL) + + expect(messages).toEqual([]) + }) + + test('deletes a single message by receipt handle', async () => { + let captured: unknown + const send = vi.fn(async (command: unknown) => { + captured = command + return {} + }) + const sqs = {send} as unknown as SQSClient + const adapter = new AwsQueueAdapter(sqs) + + await adapter.deleteMessage(QUEUE_URL, 'receipt-handle-1') + + expect(send).toHaveBeenCalledWith(expect.any(DeleteMessageCommand)) + const command = captured as DeleteMessageCommand + expect(command.input.QueueUrl).toBe(QUEUE_URL) + expect(command.input.ReceiptHandle).toBe('receipt-handle-1') + }) + + test('purges all messages from a queue', async () => { + let captured: unknown + const send = vi.fn(async (command: unknown) => { + captured = command + return {} + }) + const sqs = {send} as unknown as SQSClient + const adapter = new AwsQueueAdapter(sqs) + + await adapter.purgeQueue(QUEUE_URL) + + expect(send).toHaveBeenCalledWith(expect.any(PurgeQueueCommand)) + const command = captured as PurgeQueueCommand + expect(command.input.QueueUrl).toBe(QUEUE_URL) + }) + + test('sets MessageGroupId when sending to a FIFO queue', async () => { + let captured: unknown + const sqs = mockSqs((command) => { + if (command instanceof SendMessageCommand) { + captured = command + return {MessageId: 'msg-fifo', MD5OfMessageBody: 'def'} + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + await adapter.sendMessage('orders.fifo', 'hello') + + const command = captured as SendMessageCommand + expect(command.input.MessageGroupId).toBe('orders') + }) + + test('omits MessageGroupId when sending to a standard queue', async () => { + let captured: unknown + const sqs = mockSqs((command) => { + if (command instanceof SendMessageCommand) { + captured = command + return {MessageId: 'msg-std', MD5OfMessageBody: 'def'} + } + return {} + }) + + const adapter = new AwsQueueAdapter(sqs) + await adapter.sendMessage('orders-queue', 'hello') + + const command = captured as SendMessageCommand + expect(command.input.MessageGroupId).toBeUndefined() + }) +}) diff --git a/packages/api/src/adapter-aws/AwsQueueAdapter.ts b/packages/api/src/adapter-aws/AwsQueueAdapter.ts new file mode 100644 index 0000000..10409f7 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsQueueAdapter.ts @@ -0,0 +1,201 @@ +import { + CreateQueueCommand, + DeleteMessageCommand, + DeleteQueueCommand, + GetQueueAttributesCommand, + ListQueuesCommand, + PurgeQueueCommand, + ReceiveMessageCommand, + SendMessageCommand, + type SQSClient, +} from '@aws-sdk/client-sqs' +import {sqs as defaultSqs, defaultAccountId} from '../aws' +import {awsQueueSchema} from '../cloud-spi/queueSchema' +import type { + CloudResource, + CloudServiceAdapter, + CreateResourceInput, + QueueMessage, + ResourceQuery, + SendMessageResult, + ServiceSchema, +} from '../cloud-spi/types' + +export class AwsQueueAdapter implements CloudServiceAdapter { + readonly cloud = 'aws' as const + readonly service = 'queue' as const + + constructor(private readonly sqs: SQSClient = defaultSqs) {} + + schema(): ServiceSchema { + return awsQueueSchema() + } + + async list(query: ResourceQuery = {}): Promise { + const res = await this.sqs.send(new ListQueuesCommand({})) + const resources = (res.QueueUrls ?? []).map((url) => toResource(url)) + + return filterBySearch(resources, query.search) + } + + async get(id: string): Promise { + const queueUrl = resolveQueueUrl(id) + const base = toResource(queueUrl) + try { + const res = await this.sqs.send( + new GetQueueAttributesCommand({ + QueueUrl: queueUrl, + AttributeNames: ['CreatedTimestamp', 'ApproximateNumberOfMessages'], + }), + ) + const attributes = res.Attributes ?? {} + const createdAt = attributes.CreatedTimestamp + ? new Date(Number(attributes.CreatedTimestamp) * 1000).toISOString() + : null + return { + ...base, + createdAt, + metadata: { + ...base.metadata, + approximateNumberOfMessages: attributes.ApproximateNumberOfMessages ?? null, + }, + } + } catch { + return base + } + } + + async create(input: CreateResourceInput): Promise { + let queueName = stringValue(input.values.queueName) + if (!queueName) throw new Error('queueName is required') + if (!/^[a-zA-Z0-9_-]{1,75}(\.fifo)?$/.test(queueName)) { + throw new Error('Use a valid SQS queue name: 1-80 characters using letters, numbers, hyphens, and underscores. FIFO queues must end with .fifo.') + } + + // SQS requires FIFO queue names to end with .fifo. Normalize the name so + // selecting the FIFO option without typing the suffix still works. + if (stringValue(input.values.fifoQueue) === 'true' && !queueName.endsWith('.fifo')) { + queueName = `${queueName}.fifo` + } + + const attributes = collectAttributes(input.values) + + const res = await this.sqs.send( + new CreateQueueCommand({ + QueueName: queueName, + Attributes: Object.keys(attributes).length ? attributes : undefined, + }), + ) + + return toResource(res.QueueUrl ?? '') + } + + async delete(id: string): Promise { + await this.sqs.send(new DeleteQueueCommand({QueueUrl: resolveQueueUrl(id)})) + } + + async sendMessage(id: string, body: string): Promise { + // SQS requires MessageGroupId on every send to a FIFO queue. Derive a + // default from the queue name so FIFO sends work without a UI field. + const isFifo = id.endsWith('.fifo') + const res = await this.sqs.send( + new SendMessageCommand({ + QueueUrl: resolveQueueUrl(id), + MessageBody: body, + ...(isFifo ? {MessageGroupId: id.replace(/\.fifo$/, '')} : {}), + }), + ) + return { + messageId: res.MessageId ?? '', + md5OfMessageBody: res.MD5OfMessageBody, + } + } + + async receiveMessages(id: string, maxMessages = 10): Promise { + const res = await this.sqs.send( + new ReceiveMessageCommand({ + QueueUrl: resolveQueueUrl(id), + MaxNumberOfMessages: Math.min(Math.max(maxMessages, 1), 10), + // VisibilityTimeout 0 keeps the receive a non-consuming peek: the + // messages are returned for inspection but stay available in the + // queue instead of being hidden (consumed) for the default period. + VisibilityTimeout: 0, + AttributeNames: ['All'], + MessageAttributeNames: ['All'], + }), + ) + return (res.Messages ?? []).map((message) => ({ + messageId: message.MessageId ?? '', + body: message.Body ?? '', + receiptHandle: message.ReceiptHandle ?? '', + attributes: message.Attributes, + md5OfBody: message.MD5OfBody, + })) + } + + async deleteMessage(id: string, receiptHandle: string): Promise { + await this.sqs.send( + new DeleteMessageCommand({ + QueueUrl: resolveQueueUrl(id), + ReceiptHandle: receiptHandle, + }), + ) + } + + async purgeQueue(id: string): Promise { + await this.sqs.send( + new PurgeQueueCommand({QueueUrl: resolveQueueUrl(id)}), + ) + } +} + +function toResource(queueUrl: string): CloudResource { + const name = queueUrl.split('/').pop() ?? queueUrl + return { + id: name, + name, + cloud: 'aws', + service: 'queue', + type: 'queue', + region: null, + createdAt: null, + metadata: { + provider: 'aws', + queueService: 'sqs', + queueUrl, + }, + } +} + +// The resource id is the queue name (a safe, slash-free path segment). The full +// SQS QueueUrl is reconstructed here so the adapter can call the SDK. A full URL +// is accepted as-is for backward compatibility (e.g. direct SDK-style callers). +function resolveQueueUrl(id: string): string { + if (/^https?:\/\//i.test(id)) return id + const endpoint = process.env.FLOCI_ENDPOINT ?? 'http://localhost:4566' + const account = defaultAccountId() + return `${endpoint}/${account}/${id}` +} + +function collectAttributes(values: Record): Record { + const attributes: Record = {} + const visibilityTimeout = stringValue(values.visibilityTimeout) + if (visibilityTimeout) attributes.VisibilityTimeout = visibilityTimeout + const messageRetentionPeriod = stringValue(values.messageRetentionPeriod) + if (messageRetentionPeriod) attributes.MessageRetentionPeriod = messageRetentionPeriod + const delaySeconds = stringValue(values.delaySeconds) + if (delaySeconds) attributes.DelaySeconds = delaySeconds + const fifoQueue = stringValue(values.fifoQueue) + if (fifoQueue === 'true') attributes.FifoQueue = 'true' + return attributes +} + +function stringValue(value: unknown): string { + return typeof value === 'string' ? value.trim() : '' +} + +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)) +} \ No newline at end of file diff --git a/packages/api/src/aws.ts b/packages/api/src/aws.ts index e9ae0c9..83471e3 100644 --- a/packages/api/src/aws.ts +++ b/packages/api/src/aws.ts @@ -3,6 +3,7 @@ import { LambdaClient } from "@aws-sdk/client-lambda"; import { EKSClient } from "@aws-sdk/client-eks"; import { EC2Client } from "@aws-sdk/client-ec2"; import { RDSClient } from "@aws-sdk/client-rds"; +import { SQSClient } from "@aws-sdk/client-sqs"; import { SecretsManagerClient } from "@aws-sdk/client-secrets-manager"; const endpoint = process.env.FLOCI_ENDPOINT; @@ -38,6 +39,7 @@ export type AwsClients = { eks: EKSClient; ec2: EC2Client; rds: RDSClient; + sqs: SQSClient; secretsManager: SecretsManagerClient; }; @@ -57,6 +59,7 @@ function buildClients(accountId: string): AwsClients { eks: new EKSClient(base), ec2: new EC2Client(base), rds: new RDSClient(base), + sqs: new SQSClient(base), secretsManager: new SecretsManagerClient(base), }; } @@ -88,4 +91,5 @@ export const lambda = awsClients.lambda; export const eks = awsClients.eks; export const ec2 = awsClients.ec2; export const rds = awsClients.rds; +export const sqs = awsClients.sqs; export const secretsManager = awsClients.secretsManager; diff --git a/packages/api/src/cloud-spi/queueSchema.ts b/packages/api/src/cloud-spi/queueSchema.ts new file mode 100644 index 0000000..da4f504 --- /dev/null +++ b/packages/api/src/cloud-spi/queueSchema.ts @@ -0,0 +1,89 @@ +import type {CapabilitySchema, CloudProvider, FieldSchema, ResourceActionName, ServiceSchema, TableColumnSchema} from './types' + +const queueColumns: TableColumnSchema[] = [ + {name: 'name', label: 'Name'}, + {name: 'type', label: 'Type'}, + {name: 'cloud', label: 'Cloud'}, + {name: 'region', label: 'Region'}, + {name: 'createdAt', label: 'Created At'}, +] + +const queueFilters: FieldSchema[] = [ + {name: 'search', label: 'Search', type: 'text', required: false}, +] + +const queueResourceActions: CapabilitySchema[] = [ + {name: 'list', label: 'List resources', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'create', label: 'Create resource', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'delete', label: 'Delete resource', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'inspect', label: 'Inspect resource', enabled: true, status: 'available', runtimeRequired: false}, + {name: 'send', label: 'Send message', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'receive', label: 'Receive messages', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'deleteMessage', label: 'Delete message', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'purge', label: 'Purge queue', enabled: true, status: 'available', runtimeRequired: true}, +] + +export function awsQueueSchema(): ServiceSchema { + return { + cloud: 'aws', + service: 'queue', + displayName: 'SQS Queue', + fields: [ + { + name: 'queueName', + label: 'Queue Name', + type: 'text', + required: true, + description: '1-80 characters. Letters, numbers, hyphens, and underscores. FIFO queues must end with .fifo.', + validation: { + pattern: '^[a-zA-Z0-9_-]{1,75}(\\.fifo)?$', + minLength: 1, + maxLength: 80, + message: 'Use a valid SQS queue name: 1-80 characters using letters, numbers, hyphens, and underscores. FIFO queues must end with .fifo.', + }, + }, + { + name: 'visibilityTimeout', + label: 'Visibility Timeout (seconds)', + type: 'text', + required: false, + description: 'How long a message is hidden after being received. 0-43200. Default: 30.', + }, + { + name: 'messageRetentionPeriod', + label: 'Message Retention (seconds)', + type: 'text', + required: false, + description: 'How long messages persist. 60-1209600. Default: 345600 (4 days).', + }, + { + name: 'delaySeconds', + label: 'Delivery Delay (seconds)', + type: 'text', + required: false, + description: 'Delay before messages become available. 0-900. Default: 0.', + }, + { + name: 'fifoQueue', + label: 'FIFO Queue', + type: 'select', + required: false, + options: [ + {label: 'Standard', value: 'false'}, + {label: 'FIFO', value: 'true'}, + ], + }, + ], + actions: ['list', 'create', 'delete', 'inspect', 'send', 'receive', 'deleteMessage', 'purge'], + capabilities: { + resourceActions: queueResourceActions, + }, + filters: queueFilters, + columns: queueColumns, + } +} + +export function queueSchemaFor(cloud: CloudProvider): ServiceSchema | null { + if (cloud === 'aws') return awsQueueSchema() + return null +} \ No newline at end of file diff --git a/packages/api/src/cloud-spi/types.ts b/packages/api/src/cloud-spi/types.ts index bf9da0d..1948454 100644 --- a/packages/api/src/cloud-spi/types.ts +++ b/packages/api/src/cloud-spi/types.ts @@ -1,6 +1,6 @@ export type CloudProvider = 'aws' | 'azure' | 'gcp' -export type CloudServiceType = 'storage' | 'k8s' | 'database' | 'serverless' | 'compute' | 'networking' +export type CloudServiceType = 'storage' | 'k8s' | 'database' | 'serverless' | 'compute' | 'networking' | 'queue' export type CloudAvailability = 'available' | 'coming_soon' @@ -45,8 +45,8 @@ export interface FieldSchema { options?: Array<{label: string; value: string}> } -export type ActionSchema = 'list' | 'create' | 'delete' | 'inspect' -export type ResourceActionName = 'list' | 'create' | 'delete' | 'inspect' +export type ActionSchema = 'list' | 'create' | 'delete' | 'inspect' | 'send' | 'receive' | 'deleteMessage' | 'purge' +export type ResourceActionName = 'list' | 'create' | 'delete' | 'inspect' | 'send' | 'receive' | 'deleteMessage' | 'purge' export type ObjectActionName = 'list' | 'upload' | 'download' | 'delete' | 'createFolder' | 'copy' export type CapabilityStatus = 'available' | 'blocked' | 'partial' | 'coming_soon' @@ -83,7 +83,7 @@ export interface CloudResource { name: string cloud: CloudProvider service: CloudServiceType - type: 'bucket' | 'container' | 'cluster' | 'db-instance' | 'cosmos-database' | 'instance' | 'image' | 'vpc' | 'lambda' | 'azure-function' | 'gcp-function' + type: 'bucket' | 'container' | 'cluster' | 'db-instance' | 'cosmos-database' | 'instance' | 'image' | 'vpc' | 'lambda' | 'azure-function' | 'gcp-function' | 'queue' region: string | null createdAt: string | null status?: string | null @@ -151,6 +151,18 @@ export interface ServerlessInvokeResult { logResult?: string executionDuration?: number } +export interface SendMessageResult { + messageId: string + md5OfMessageBody?: string +} + +export interface QueueMessage { + messageId: string + body: string + receiptHandle: string + attributes?: Record + md5OfBody?: string +} export interface CloudServiceAdapter { readonly cloud: CloudProvider readonly service: CloudServiceType @@ -164,6 +176,10 @@ export interface CloudServiceAdapter { getObject?(resourceId: string, key: string): Promise deleteObject?(resourceId: string, key: string): Promise invoke?(id: string, payload: string): Promise + sendMessage?(id: string, body: string): Promise + receiveMessages?(id: string, maxMessages?: number): Promise + deleteMessage?(id: string, receiptHandle: string): Promise + purgeQueue?(id: string): Promise copyObject?(srcResourceId: string, srcKey: string, destKey: string, destResourceId?: string): Promise listCosmosContainers?(databaseId: string): Promise createCosmosContainer?(databaseId: string, input: CreateResourceInput): Promise diff --git a/packages/api/src/cloudProxy.ts b/packages/api/src/cloudProxy.ts index b9740c0..2bf2e4e 100644 --- a/packages/api/src/cloudProxy.ts +++ b/packages/api/src/cloudProxy.ts @@ -4,6 +4,7 @@ import {AwsNetworkingAdapter} from './adapter-aws/AwsNetworkingAdapter' import {AwsDatabaseAdapter} from './adapter-aws/AwsDatabaseAdapter' import {AwsEksAdapter} from './adapter-aws/AwsEksAdapter' import {AwsStorageAdapter} from './adapter-aws/AwsStorageAdapter' +import {AwsQueueAdapter} from './adapter-aws/AwsQueueAdapter' import {AzureDatabaseAdapter} from './adapter-azure/AzureDatabaseAdapter' import {AzureStorageAdapter} from './adapter-azure/AzureStorageAdapter' import {GcpStorageAdapter} from './adapter-gcp/GcpStorageAdapter' @@ -28,6 +29,7 @@ export function createCloudProxyService(accountId?: string | null): CloudProxySe const registry = new CloudAdapterRegistry([ new AwsStorageAdapter(clients.s3), + new AwsQueueAdapter(clients.sqs), new AwsEksAdapter(createEksService(clients.eks)), new AwsDatabaseAdapter(createRdsService(clients.rds), clients.rds), new AwsComputeAdapter(ec2Service), diff --git a/packages/api/src/routes/clouds.ts b/packages/api/src/routes/clouds.ts index 93cb831..3a2e0b6 100644 --- a/packages/api/src/routes/clouds.ts +++ b/packages/api/src/routes/clouds.ts @@ -242,6 +242,73 @@ export function createCloudRoutes(injectedService?: CloudProxyService) { }) }) + app.post('/:cloud/services/:service/resources/:id/send', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + const serviceType = c.req.param('service') as CloudServiceType + if (!isCloudProvider(cloud) || !isServiceType(serviceType)) { + return c.json({error: 'Unknown cloud or service'}, 404) + } + + return withRuntime(c, async () => { + const body: {body?: string} = await c.req.json<{body?: string}>().catch(() => ({})) + const result = await svc(c).sendQueueMessage( + cloud, + serviceType, + c.req.param('id'), + body.body ?? '', + ) + return c.json(result, 201) + }) + }) + + app.post('/:cloud/services/:service/resources/:id/receive', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + const serviceType = c.req.param('service') as CloudServiceType + if (!isCloudProvider(cloud) || !isServiceType(serviceType)) { + return c.json({error: 'Unknown cloud or service'}, 404) + } + + return withRuntime(c, async () => { + const body: {maxMessages?: number} = await c.req.json<{maxMessages?: number}>().catch(() => ({})) + const messages = await svc(c).receiveQueueMessages( + cloud, + serviceType, + c.req.param('id'), + body.maxMessages, + ) + return c.json(messages, 200) + }) + }) + + app.delete('/:cloud/services/:service/resources/:id/messages', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + const serviceType = c.req.param('service') as CloudServiceType + if (!isCloudProvider(cloud) || !isServiceType(serviceType)) { + return c.json({error: 'Unknown cloud or service'}, 404) + } + + const receiptHandle = c.req.query('receiptHandle') + if (!receiptHandle) return c.json({error: 'receiptHandle is required'}, 400) + + return withRuntime(c, async () => { + await svc(c).deleteQueueMessage(cloud, serviceType, c.req.param('id'), receiptHandle) + return c.json({ok: true}) + }) + }) + + app.post('/:cloud/services/:service/resources/:id/purge', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + const serviceType = c.req.param('service') as CloudServiceType + if (!isCloudProvider(cloud) || !isServiceType(serviceType)) { + return c.json({error: 'Unknown cloud or service'}, 404) + } + + return withRuntime(c, async () => { + await svc(c).purgeQueue(cloud, serviceType, c.req.param('id')) + return c.json({ok: true}) + }) + }) + app.delete('/:cloud/services/:service/resources/:id', async (c) => { const cloud = c.req.param('cloud') as CloudProvider const serviceType = c.req.param('service') as CloudServiceType @@ -261,7 +328,7 @@ function isCloudProvider(value: string): value is CloudProvider { } function isServiceType(value: string): value is CloudServiceType { - return value === 'storage' || value === 'k8s' || value === 'database' || value === 'serverless' || value === 'compute' || value === 'networking' + return value === 'storage' || value === 'k8s' || value === 'database' || value === 'serverless' || value === 'compute' || value === 'networking' || value === 'queue' } async function withRuntime(c: Context, handler: () => Promise): Promise { diff --git a/packages/api/src/service/CloudProxyService.ts b/packages/api/src/service/CloudProxyService.ts index 826bf12..d9cf9e5 100644 --- a/packages/api/src/service/CloudProxyService.ts +++ b/packages/api/src/service/CloudProxyService.ts @@ -10,16 +10,19 @@ import type { CosmosQueryResult, CreateResourceInput, ResourceQuery, + SendMessageResult, ServerlessInvokeResult, ServiceSchema, StorageObjectDownload, StorageObjectList, + QueueMessage, } from '../cloud-spi/types' import {storageSchemaFor} from '../cloud-spi/storageSchema' import {CloudAdapterRegistry} from '../registry/CloudAdapterRegistry' import {serverlessSchemaFor} from '../cloud-spi/serverlessSchema' import {k8sSchemaFor} from '../cloud-spi/eksSchema' import {databaseSchemaFor} from '../cloud-spi/databaseSchema' +import {queueSchemaFor} from '../cloud-spi/queueSchema' import {azureEndpoint} from '../azure' import {checkGcpRuntime, gcpEndpoint} from '../gcp' @@ -73,6 +76,12 @@ export class CloudProxyService { displayName: 'Networking', availability: this.registry.get(cloud, 'networking') ? 'available' : 'coming_soon', }) + services.push({ + cloud, + service: 'queue', + displayName: 'Queue', + availability: this.registry.get(cloud, 'queue') ? 'available' : 'coming_soon', + }) return services } @@ -83,6 +92,7 @@ export class CloudProxyService { if (service === 'k8s') return k8sSchemaFor(cloud) if (service === 'database') return databaseSchemaFor(cloud) if (service === 'serverless') return serverlessSchemaFor(cloud) + if (service === 'queue') return queueSchemaFor(cloud) return null } @@ -169,6 +179,48 @@ async invokeResource( if (!adapter.invoke) throw new Error(`${cloud}/${service} invoke is not supported`) return adapter.invoke(id, payload) } + async sendQueueMessage( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + body: string, + ): Promise { + const adapter = this.requireAdapter(cloud, service) + if (!adapter.sendMessage) throw new Error(`${cloud}/${service} send message is not supported`) + return adapter.sendMessage(id, body) + } + + async receiveQueueMessages( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + maxMessages?: number, + ): Promise { + const adapter = this.requireAdapter(cloud, service) + if (!adapter.receiveMessages) throw new Error(`${cloud}/${service} receive messages is not supported`) + return adapter.receiveMessages(id, maxMessages) + } + + async deleteQueueMessage( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + receiptHandle: string, + ): Promise { + const adapter = this.requireAdapter(cloud, service) + if (!adapter.deleteMessage) throw new Error(`${cloud}/${service} delete message is not supported`) + return adapter.deleteMessage(id, receiptHandle) + } + + async purgeQueue( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + ): Promise { + const adapter = this.requireAdapter(cloud, service) + if (!adapter.purgeQueue) throw new Error(`${cloud}/${service} purge queue is not supported`) + return adapter.purgeQueue(id) + } async listObjects(cloud: CloudProvider, service: CloudServiceType, resourceId: string, prefix?: string): Promise { const adapter = this.requireAdapter(cloud, service) if (!adapter.listObjects) throw new Error(`Object listing is not supported for ${cloud}/${service}`) diff --git a/packages/frontend/src/api/api.ts b/packages/frontend/src/api/api.ts index a45b849..f07feec 100644 --- a/packages/frontend/src/api/api.ts +++ b/packages/frontend/src/api/api.ts @@ -14,13 +14,17 @@ export const apiEndpointKeys = { services: "clouds.services.list", status: "clouds.status.get", schema: "clouds.services.schema.get", - resources: { - list: "clouds.services.resources.list", - get: "clouds.services.resources.get", - create: "clouds.services.resources.create", - delete: "clouds.services.resources.delete", - invoke: "clouds.services.resources.invoke", - }, + resources: { + list: "clouds.services.resources.list", + get: "clouds.services.resources.get", + create: "clouds.services.resources.create", + delete: "clouds.services.resources.delete", + invoke: "clouds.services.resources.invoke", + send: "clouds.services.resources.send", + receive: "clouds.services.resources.receive", + deleteMessage: "clouds.services.resources.messages.delete", + purge: "clouds.services.resources.purge", + }, storage: { objects: { list: "clouds.services.storage.objects.list", @@ -211,13 +215,45 @@ export const endpointRegistry: EndpointRegistry = new Map([ }, ], [ - apiEndpointKeys.clouds.resources.invoke, - { - path: "/clouds/:cloud/services/:service/resources/:id/invoke", - method: "POST", - telemetry: { service: "cloud-proxy" }, - }, -], + apiEndpointKeys.clouds.resources.invoke, + { + path: "/clouds/:cloud/services/:service/resources/:id/invoke", + method: "POST", + telemetry: { service: "cloud-proxy" }, + }, + ], + [ + apiEndpointKeys.clouds.resources.send, + { + path: "/clouds/:cloud/services/:service/resources/:id/send", + method: "POST", + telemetry: { service: "cloud-proxy" }, + }, + ], + [ + apiEndpointKeys.clouds.resources.receive, + { + path: "/clouds/:cloud/services/:service/resources/:id/receive", + method: "POST", + telemetry: { service: "cloud-proxy" }, + }, + ], + [ + apiEndpointKeys.clouds.resources.deleteMessage, + { + path: "/clouds/:cloud/services/:service/resources/:id/messages", + method: "DELETE", + telemetry: { service: "cloud-proxy" }, + }, + ], + [ + apiEndpointKeys.clouds.resources.purge, + { + path: "/clouds/:cloud/services/:service/resources/:id/purge", + method: "POST", + telemetry: { service: "cloud-proxy" }, + }, + ], [ apiEndpointKeys.clouds.resources.get, { @@ -243,15 +279,6 @@ export const endpointRegistry: EndpointRegistry = new Map([ }, ], -[ - apiEndpointKeys.clouds.resources.invoke, - { - path: "/clouds/:cloud/services/:service/resources/:id/invoke", - method: "POST", - telemetry: { service: "cloud-proxy" }, - }, -], - [ apiEndpointKeys.clouds.storage.objects.list, { diff --git a/packages/frontend/src/api/cloudProxyClient.ts b/packages/frontend/src/api/cloudProxyClient.ts index 9bd9fc7..a61fcc1 100644 --- a/packages/frontend/src/api/cloudProxyClient.ts +++ b/packages/frontend/src/api/cloudProxyClient.ts @@ -6,7 +6,7 @@ import type { CloudServiceType, CloudStatus, } from "@/types/cloud"; -import type { CloudResource, CosmosContainer, CosmosItem, CosmosQueryResult, StorageObjectList } from "@/types/resource"; +import type { CloudResource, CosmosContainer, CosmosItem, CosmosQueryResult, QueueMessage, StorageObjectList } from "@/types/resource"; import type { ServiceSchema } from "@/types/schema"; import { getAccountId } from "@/lib/accountStore"; @@ -113,14 +113,74 @@ export async function deleteCloudResource( { cloud, service, id }, ); } -export interface ServerlessInvokeResult { - statusCode: number; - payload: string; - functionError?: string; - logResult?: string; - executionDuration?: number; + +export interface SendMessageResult { + messageId: string + md5OfMessageBody?: string } +export async function sendQueueMessage( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + body: string, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.resources.send, + requestOptions(cloud, service, { signal, body: { body } }), + { cloud, service, id }, + ); + return res.data; +} + +export async function receiveQueueMessages( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + maxMessages?: number, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.resources.receive, + requestOptions(cloud, service, { + signal, + body: maxMessages ? { maxMessages } : {}, + }), + { cloud, service, id }, + ); + return res.data; +} + +export async function deleteQueueMessage( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + receiptHandle: string, + signal?: AbortSignal, +): Promise { + await apiClient.call( + apiEndpointKeys.clouds.resources.deleteMessage, + requestOptions(cloud, service, { + signal, + params: { receiptHandle }, + }), + { cloud, service, id }, + ); +} + +export async function purgeQueue( + cloud: CloudProvider, + service: CloudServiceType, + id: string, + signal?: AbortSignal, +): Promise { + await apiClient.call( + apiEndpointKeys.clouds.resources.purge, + requestOptions(cloud, service, { signal }), + { cloud, service, id }, + ); +} export interface ServerlessInvokeResult { statusCode: number; payload: string; diff --git a/packages/frontend/src/components/DynamicResourceView.tsx b/packages/frontend/src/components/DynamicResourceView.tsx index 706b77b..f16e959 100644 --- a/packages/frontend/src/components/DynamicResourceView.tsx +++ b/packages/frontend/src/components/DynamicResourceView.tsx @@ -31,6 +31,7 @@ import type { CloudResource, StorageObject } from "@/types/resource"; import type { ServiceSchema } from "@/types/schema"; import { CosmosNoSqlPanel } from "@/components/CosmosNoSqlPanel"; import { ServerlessInvokePanel } from "@/components/ServerlessInvokePanel"; +import { QueuePanel } from "@/components/QueuePanel"; interface DynamicResourceViewProps { cloud: CloudProvider; @@ -325,6 +326,13 @@ export function DynamicResourceView({ runtimeReachable={canUseRuntime} /> )} + {service === "queue" && ( + + )} ); } diff --git a/packages/frontend/src/components/Layout.tsx b/packages/frontend/src/components/Layout.tsx index 31a1891..2a10d65 100644 --- a/packages/frontend/src/components/Layout.tsx +++ b/packages/frontend/src/components/Layout.tsx @@ -52,7 +52,7 @@ const CLOUD_SERVICE_ITEMS: Array<{name: CloudSidebarService; label: string; rout {name: 'networking', label: 'Networking', route: 'networking'}, {name: 'secretsmanager', label: 'Secrets Manager', route: '/secretsmanager'}, {name: 'serverless', label: 'Serverless', route: 'serverless'}, - {name: 'queue', label: 'Queue'}, + {name: 'queue', label: 'Queue', route: 'queue'}, {name: 'function', label: 'Function'}, ] @@ -71,6 +71,7 @@ function CloudServiceNav() { || (service.name === 'database' && (cloud === 'aws' || cloud === 'azure')) || ((service.name === 'k8s' || service.name === 'compute' || service.name === 'networking') && cloud === 'aws') || (service.name === 'serverless' && (cloud === 'aws' || cloud === 'azure')) + || (service.name === 'queue' && cloud === 'aws') if (service.route && available) { const target = service.route.startsWith('/') ? service.route : `/cloud-explorer/${cloud}/${service.route}` return diff --git a/packages/frontend/src/components/QueuePanel.tsx b/packages/frontend/src/components/QueuePanel.tsx new file mode 100644 index 0000000..94ab1b3 --- /dev/null +++ b/packages/frontend/src/components/QueuePanel.tsx @@ -0,0 +1,270 @@ +import {useEffect, useState} from "react"; +import {Inbox, Loader2, MessageSquare, RefreshCw, Send, Trash2} from "lucide-react"; +import {useMutation, useQuery} from "@tanstack/react-query"; +import { + deleteQueueMessage, + purgeQueue, + receiveQueueMessages, + sendQueueMessage, + type SendMessageResult, +} from "@/api/cloudProxyClient"; +import type {CloudProvider} from "@/types/cloud"; +import type {CloudResource, QueueMessage} from "@/types/resource"; + +interface QueuePanelProps { + cloud: CloudProvider; + resource?: CloudResource; + runtimeReachable: boolean; +} + +export function QueuePanel({cloud, resource, runtimeReachable}: QueuePanelProps) { + const [messageBody, setMessageBody] = useState(""); + const [sendResult, setSendResult] = useState(null); + const [sendError, setSendError] = useState(null); + const [opError, setOpError] = useState(null); + const [deletingHandles, setDeletingHandles] = useState>(new Set()); + + useEffect(() => { + setMessageBody(""); + setSendResult(null); + setSendError(null); + setOpError(null); + }, [resource?.id]); + + const isQueue = resource?.service === "queue" && resource.type === "queue"; + const canAct = Boolean(resource && isQueue && runtimeReachable); + + const sendMutation = useMutation({ + mutationFn: () => + sendQueueMessage(cloud, "queue", resource!.id, messageBody), + onSuccess: (result) => { + setSendResult(result); + setSendError(null); + setMessageBody(""); + void messagesQuery.refetch(); + }, + onError: (error) => { + setSendError(error instanceof Error ? error.message : "Failed to send message"); + }, + }); + + const messagesQuery = useQuery({ + queryKey: ["queue-messages", cloud, resource?.id], + queryFn: ({signal}) => + receiveQueueMessages(cloud, "queue", resource!.id, 10, signal), + enabled: canAct, + }); + + const deleteMessageMutation = useMutation({ + mutationFn: (receiptHandle: string) => + deleteQueueMessage(cloud, "queue", resource!.id, receiptHandle), + onMutate: (receiptHandle) => { + setDeletingHandles((prev) => new Set(prev).add(receiptHandle)); + }, + onSettled: (_result, _error, receiptHandle) => { + setDeletingHandles((prev) => { + const next = new Set(prev); + next.delete(receiptHandle); + return next; + }); + }, + onSuccess: () => { + setOpError(null); + void messagesQuery.refetch(); + }, + onError: (error) => { + setOpError(error instanceof Error ? error.message : "Failed to delete message"); + }, + }); + + const purgeMutation = useMutation({ + mutationFn: () => purgeQueue(cloud, "queue", resource!.id), + onSuccess: () => { + setOpError(null); + void messagesQuery.refetch(); + }, + onError: (error) => { + setOpError(error instanceof Error ? error.message : "Failed to purge queue"); + }, + }); + + if (!resource || resource.service !== "queue") { + return ( +
+
+

Select a queue

+

+ Select an SQS queue to send a message and view the messages it + currently holds. +

+
+
+ ); + } + + return ( +
+
+
+

Queue Actions

+

+ + {resource.name} +

+

+ Send a message to this SQS queue and receive the messages it + currently holds. +

+
+ + {canAct ? "Ready" : "Runtime unavailable"} + +
+ +
+
+ +
+