Feat/adding UI for aws sqs

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"
   },

packages/api/src/adapter-aws/AwsQueueAdapter.test.ts

@@ -0,0 +1,119 @@
+import {describe, expect, test} from 'bun:test'
+import {AwsQueueAdapter} from './AwsQueueAdapter'
+
+const QUEUE_URL = 'http://localhost:4566/000000000000/orders'
+
+function fakeSqs(overrides: Record<string, (input: unknown) => unknown> = {}) {
+    return {
+        async send(command: {constructor: {name: string}; input: unknown}) {
+            const handler = overrides[command.constructor.name]
+            if (handler) return handler(command.input)
+
+            switch (command.constructor.name) {
+                case 'ListQueuesCommand':
+                    return {QueueUrls: [QUEUE_URL]}
+                case 'GetQueueAttributesCommand':
+                    return {
+                        Attributes: {
+                            QueueArn: 'arn:aws:sqs:us-east-1:000000000000:orders',
+                            ApproximateNumberOfMessages: '3',
+                            ApproximateNumberOfMessagesNotVisible: '1',
+                            ApproximateNumberOfMessagesDelayed: '0',
+                            CreatedTimestamp: '1700000000',
+                            FifoQueue: 'false',
+                        },
+                    }
+                default:
+                    throw new Error(`Unhandled command: ${command.constructor.name}`)
+            }
+        },
+    } as never
+}
+
+describe('AwsQueueAdapter', () => {
+    test('list returns mapped CloudResource array with message counts in metadata', async () => {
+        const adapter = new AwsQueueAdapter(fakeSqs())
+        const result = await adapter.list()
+
+        expect(result).toHaveLength(1)
+        expect(result[0].id).toBe(QUEUE_URL)
+        expect(result[0].name).toBe('orders')
+        expect(result[0].cloud).toBe('aws')
+        expect(result[0].service).toBe('queue')
+        expect(result[0].type).toBe('queue')
+        expect(result[0].metadata.messagesAvailable).toBe(3)
+        expect(result[0].metadata.messagesInFlight).toBe(1)
+    })
+
+    test('list filters results by search term', async () => {
+        const adapter = new AwsQueueAdapter(fakeSqs({
+            ListQueuesCommand: () => ({QueueUrls: [QUEUE_URL, `${QUEUE_URL}-dlq`]}),
+        }))
+
+        const result = await adapter.list({search: 'orders-dlq'})
+
+        expect(result).toHaveLength(1)
+        expect(result[0].name).toBe('orders-dlq')
+    })
+
+    test('create builds a queue and returns the mapped resource', async () => {
+        const adapter = new AwsQueueAdapter(fakeSqs({
+            CreateQueueCommand: () => ({QueueUrl: QUEUE_URL}),
+        }))
+
+        const resource = await adapter.create({values: {queueName: 'orders'}})
+
+        expect(resource.id).toBe(QUEUE_URL)
+        expect(resource.name).toBe('orders')
+    })
+
+    test('create rejects a missing queueName', async () => {
+        const adapter = new AwsQueueAdapter(fakeSqs())
+        await expect(adapter.create({values: {}})).rejects.toThrow('queueName is required')
+    })
+
+    test('sendMessage returns the message id from the SDK response', async () => {
+        const adapter = new AwsQueueAdapter(fakeSqs({
+            SendMessageCommand: () => ({MessageId: 'msg-1'}),
+        }))
+
+        const message = await adapter.sendMessage(QUEUE_URL, '{"hello":"world"}')
+
+        expect(message.id).toBe('msg-1')
+        expect(message.body).toBe('{"hello":"world"}')
+    })
+
+    test('receiveMessages maps SDK messages into QueueMessage records', async () => {
+        const adapter = new AwsQueueAdapter(fakeSqs({
+            ReceiveMessageCommand: () => ({
+                Messages: [
+                    {
+                        MessageId: 'msg-1',
+                        ReceiptHandle: 'receipt-1',
+                        Body: '{"hello":"world"}',
+                        Attributes: {SentTimestamp: '1700000000000', ApproximateReceiveCount: '2'},
+                        MessageAttributes: {source: {StringValue: 'flow-panel', DataType: 'String'}},
+                    },
+                ],
+            }),
+        }))
+
+        const messages = await adapter.receiveMessages(QUEUE_URL)
+
+        expect(messages).toHaveLength(1)
+        expect(messages[0].id).toBe('msg-1')
+        expect(messages[0].receiptHandle).toBe('receipt-1')
+        expect(messages[0].receiveCount).toBe(2)
+        expect(messages[0].messageAttributes.source).toBe('flow-panel')
+    })
+
+    test('deleteMessage and purgeQueue delegate to the SDK without throwing', async () => {
+        const adapter = new AwsQueueAdapter(fakeSqs({
+            DeleteMessageCommand: () => ({}),
+            PurgeQueueCommand: () => ({}),
+        }))
+
+        await expect(adapter.deleteMessage(QUEUE_URL, 'receipt-1')).resolves.toBeUndefined()
+        await expect(adapter.purgeQueue(QUEUE_URL)).resolves.toBeUndefined()
+    })
+})

packages/api/src/adapter-aws/AwsQueueAdapter.ts

@@ -0,0 +1,216 @@
+import {
+    CreateQueueCommand,
+    DeleteMessageCommand,
+    DeleteQueueCommand,
+    GetQueueAttributesCommand,
+    ListQueuesCommand,
+    PurgeQueueCommand,
+    QueueAttributeName,
+    ReceiveMessageCommand,
+    SendMessageCommand,
+    type SQSClient,
+} from '@aws-sdk/client-sqs'
+import {awsQueueSchema} from '../cloud-spi/queueSchema'
+import type {
+    CloudResource,
+    CloudServiceAdapter,
+    CreateResourceInput,
+    QueueMessage,
+    ResourceQuery,
+    ServiceSchema,
+} from '../cloud-spi/types'
+import {sqs as defaultSqs} from '../aws'
+
+const ATTRIBUTES_FOR_LIST: QueueAttributeName[] = [
+    QueueAttributeName.QueueArn,
+    QueueAttributeName.ApproximateNumberOfMessages,
+    QueueAttributeName.ApproximateNumberOfMessagesNotVisible,
+    QueueAttributeName.ApproximateNumberOfMessagesDelayed,
+    QueueAttributeName.CreatedTimestamp,
+    QueueAttributeName.VisibilityTimeout,
+    QueueAttributeName.DelaySeconds,
+    QueueAttributeName.MessageRetentionPeriod,
+    QueueAttributeName.RedrivePolicy,
+    QueueAttributeName.FifoQueue,
+]
+
+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<CloudResource[]> {
+        const res = await this.sqs.send(new ListQueuesCommand({}))
+        const urls = res.QueueUrls ?? []
+
+        const resources = await Promise.all(
+            urls.map((url) => this.toResource(url)),
+        )
+
+        return filterBySearch(resources, query.search)
+    }
+
+    async get(id: string): Promise<CloudResource | null> {
+        try {
+            return await this.toResource(id)
+        } catch (error) {
+            if (isQueueMissing(error)) return null
+            throw error
+        }
+    }
+
+    async create(input: CreateResourceInput): Promise<CloudResource> {
+        const values = input.values
+        const queueName = stringValue(values.queueName)
+        if (!queueName) throw new Error('queueName is required')
+
+        const fifo = String(values.fifo ?? '') === 'fifo' || queueName.endsWith('.fifo')
+        const name = fifo && !queueName.endsWith('.fifo') ? `${queueName}.fifo` : queueName
+
+        const attributes: Record<string, string> = {}
+        if (fifo) attributes.FifoQueue = 'true'
+        const visibilityTimeout = stringValue(values.visibilityTimeout)
+        if (visibilityTimeout) attributes.VisibilityTimeout = visibilityTimeout
+        const delaySeconds = stringValue(values.delaySeconds)
+        if (delaySeconds) attributes.DelaySeconds = delaySeconds
+        const messageRetentionPeriod = stringValue(values.messageRetentionPeriod)
+        if (messageRetentionPeriod) attributes.MessageRetentionPeriod = messageRetentionPeriod
+
+        const res = await this.sqs.send(new CreateQueueCommand({
+            QueueName: name,
+            Attributes: Object.keys(attributes).length ? attributes : undefined,
+        }))
+
+        if (!res.QueueUrl) throw new Error('Floci runtime did not return a queue URL')
+        return this.toResource(res.QueueUrl)
+    }
+
+    async delete(id: string): Promise<void> {
+        await this.sqs.send(new DeleteQueueCommand({QueueUrl: id}))
+    }
+
+    async sendMessage(queueId: string, body: string, messageAttributes?: Record<string, string>): Promise<QueueMessage> {
+        const res = await this.sqs.send(new SendMessageCommand({
+            QueueUrl: queueId,
+            MessageBody: body,
+            MessageAttributes: messageAttributes
+                ? Object.fromEntries(
+                    Object.entries(messageAttributes).map(([key, value]) => [
+                        key,
+                        {DataType: 'String', StringValue: value},
+                    ]),
+                )
+                : undefined,
+        }))
+
+        return {
+            id: res.MessageId ?? '',
+            receiptHandle: '',
+            body,
+            attributes: {},
+            messageAttributes: messageAttributes ?? {},
+            sentAt: new Date().toISOString(),
+            receiveCount: null,
+        }
+    }
+
+    async receiveMessages(queueId: string, maxMessages = 5, waitTimeSeconds = 0): Promise<QueueMessage[]> {
+        const res = await this.sqs.send(new ReceiveMessageCommand({
+            QueueUrl: queueId,
+            MaxNumberOfMessages: clamp(maxMessages, 1, 10),
+            WaitTimeSeconds: clamp(waitTimeSeconds, 0, 20),
+            MessageAttributeNames: ['All'],
+            AttributeNames: [QueueAttributeName.All],
+        }))
+
+        return (res.Messages ?? []).map((message) => ({
+            id: message.MessageId ?? '',
+            receiptHandle: message.ReceiptHandle ?? '',
+            body: message.Body ?? '',
+            attributes: message.Attributes ?? {},
+            messageAttributes: Object.fromEntries(
+                Object.entries(message.MessageAttributes ?? {}).map(([key, value]) => [key, value.StringValue ?? '']),
+            ),
+            sentAt: message.Attributes?.SentTimestamp
+                ? new Date(Number(message.Attributes.SentTimestamp)).toISOString()
+                : null,
+            receiveCount: message.Attributes?.ApproximateReceiveCount
+                ? Number(message.Attributes.ApproximateReceiveCount)
+                : null,
+        }))
+    }
+
+    async deleteMessage(queueId: string, receiptHandle: string): Promise<void> {
+        await this.sqs.send(new DeleteMessageCommand({QueueUrl: queueId, ReceiptHandle: receiptHandle}))
+    }
+
+    async purgeQueue(queueId: string): Promise<void> {
+        await this.sqs.send(new PurgeQueueCommand({QueueUrl: queueId}))
+    }
+
+    private async toResource(queueUrl: string): Promise<CloudResource> {
+        const res = await this.sqs.send(new GetQueueAttributesCommand({
+            QueueUrl: queueUrl,
+            AttributeNames: ATTRIBUTES_FOR_LIST,
+        }))
+        const attrs = res.Attributes ?? {}
+        const name = queueUrl.split('/').pop() ?? queueUrl
+
+        return {
+            id: queueUrl,
+            name,
+            cloud: 'aws',
+            service: 'queue',
+            type: 'queue',
+            region: null,
+            createdAt: attrs.CreatedTimestamp
+                ? new Date(Number(attrs.CreatedTimestamp) * 1000).toISOString()
+                : null,
+            status: 'Active',
+            metadata: {
+                queueUrl,
+                queueArn: attrs.QueueArn,
+                fifo: attrs.FifoQueue === 'true',
+                messagesAvailable: numberOr(attrs.ApproximateNumberOfMessages, 0),
+                messagesInFlight: numberOr(attrs.ApproximateNumberOfMessagesNotVisible, 0),
+                messagesDelayed: numberOr(attrs.ApproximateNumberOfMessagesDelayed, 0),
+                visibilityTimeout: numberOr(attrs.VisibilityTimeout, null),
+                delaySeconds: numberOr(attrs.DelaySeconds, null),
+                messageRetentionPeriod: numberOr(attrs.MessageRetentionPeriod, null),
+                redrivePolicy: attrs.RedrivePolicy ? JSON.parse(attrs.RedrivePolicy) : null,
+            },
+        }
+    }
+}
+
+function numberOr(value: string | undefined, fallback: number | null): number | null {
+    if (value === undefined) return fallback
+    const parsed = Number(value)
+    return Number.isFinite(parsed) ? parsed : fallback
+}
+
+function clamp(value: number, min: number, max: number): number {
+    if (!Number.isFinite(value)) return min
+    return Math.min(max, Math.max(min, value))
+}
+
+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((r) => r.name.toLowerCase().includes(normalized))
+}
+
+function isQueueMissing(error: unknown): boolean {
+    if (typeof error !== 'object' || error === null) return false
+    const name = (error as {name?: string}).name
+    return name === 'QueueDoesNotExist' || name === 'AWS.SimpleQueueService.NonExistentQueue'
+}

packages/api/src/aws.ts

@@ -4,6 +4,7 @@ import { EKSClient } from "@aws-sdk/client-eks";
 import { EC2Client } from "@aws-sdk/client-ec2";
 import { RDSClient } from "@aws-sdk/client-rds";
 import { SecretsManagerClient } from "@aws-sdk/client-secrets-manager";
+import { SQSClient } from "@aws-sdk/client-sqs";
 
 const endpoint = process.env.FLOCI_ENDPOINT;
 const region = process.env.AWS_REGION || "us-east-1";
@@ -39,6 +40,7 @@ export type AwsClients = {
   ec2: EC2Client;
   rds: RDSClient;
   secretsManager: SecretsManagerClient;
+  sqs: SQSClient;
 };
 
 export type AwsClientName = keyof AwsClients;
@@ -58,6 +60,7 @@ function buildClients(accountId: string): AwsClients {
     ec2: new EC2Client(base),
     rds: new RDSClient(base),
     secretsManager: new SecretsManagerClient(base),
+    sqs: new SQSClient(base),
   };
 }
 
@@ -89,3 +92,4 @@ export const eks = awsClients.eks;
 export const ec2 = awsClients.ec2;
 export const rds = awsClients.rds;
 export const secretsManager = awsClients.secretsManager;
+export const sqs = awsClients.sqs;