feat: Cloudformation + RDS

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

@@ -1,9 +1,15 @@
 import {
     CreateQueueCommand,
+    DeleteMessageCommand,
     DeleteQueueCommand,
     GetQueueAttributesCommand,
     GetQueueUrlCommand,
     ListQueuesCommand,
+    PurgeQueueCommand,
+    ReceiveMessageCommand,
+    SendMessageCommand,
+    SetQueueAttributesCommand,
+    StartMessageMoveTaskCommand,
     type SQSClient,
 } from '@aws-sdk/client-sqs'
 import {NotFoundError, ValidationError} from '../cloud-spi/errors'
@@ -15,6 +21,12 @@ import type {
     CreateResourceInput,
     ResourceQuery,
     ServiceSchema,
+    SqsMessage,
+    SqsMoveTaskInput,
+    SqsMoveTaskResult,
+    SqsReceiveMessagesInput,
+    SqsSendMessageInput,
+    UpdateResourceInput,
 } from '../cloud-spi/types'
 
 /**
@@ -38,6 +50,7 @@ interface QueueAttributes {
     CreatedTimestamp?: string
     LastModifiedTimestamp?: string
     FifoQueue?: string
+    RedrivePolicy?: string
 }
 
 export class AwsSqsAdapter implements CloudServiceAdapter {
@@ -111,6 +124,69 @@ export class AwsSqsAdapter implements CloudServiceAdapter {
         await this.sqs.send(new DeleteQueueCommand({QueueUrl: url}))
     }
 
+    async update(id: string, input: UpdateResourceInput): Promise<CloudResource> {
+        const url = await this.requireQueueUrl(id)
+        const attributes: Record<string, string> = {}
+        const visibility = numericAttribute(input.values.visibilityTimeout, 'visibilityTimeout', 0, 43_200)
+        const retention = numericAttribute(input.values.messageRetentionPeriod, 'messageRetentionPeriod', 60, 1_209_600)
+        if (visibility) attributes.VisibilityTimeout = visibility
+        if (retention) attributes.MessageRetentionPeriod = retention
+        const redrive = input.values.redrivePolicy
+        if (redrive !== undefined) {
+            if (!redrive || typeof redrive !== 'object' || Array.isArray(redrive)) throw new ValidationError('redrivePolicy must be an object.')
+            const policy = redrive as {deadLetterTargetArn?: unknown; maxReceiveCount?: unknown}
+            if (typeof policy.deadLetterTargetArn !== 'string' || !policy.deadLetterTargetArn) throw new ValidationError('A dead-letter queue is required.')
+            const count = numericAttribute(policy.maxReceiveCount, 'maxReceiveCount', 1, 1_000)
+            if (!count) throw new ValidationError('maxReceiveCount is required.')
+            attributes.RedrivePolicy = JSON.stringify({deadLetterTargetArn: policy.deadLetterTargetArn, maxReceiveCount: count})
+        }
+        if (Object.keys(attributes).length === 0) throw new ValidationError('No SQS queue settings were supplied.')
+        await this.sqs.send(new SetQueueAttributesCommand({QueueUrl: url, Attributes: attributes}))
+        return (await this.get(id))!
+    }
+
+    async sendSqsMessage(queueId: string, input: SqsSendMessageInput): Promise<{messageId: string}> {
+        const url = await this.requireQueueUrl(queueId)
+        if (!input.messageBody) throw new ValidationError('messageBody is required.')
+        const queue = await this.get(queueId)
+        if (queue?.type === 'fifo-queue' && !input.messageGroupId) throw new ValidationError('messageGroupId is required for FIFO queues.')
+        const messageAttributes = Object.fromEntries(Object.entries(input.messageAttributes ?? {}).map(([name, value]) => [name, {
+            DataType: value.dataType,
+            ...(value.stringValue !== undefined ? {StringValue: value.stringValue} : {}),
+            ...(value.binaryValue !== undefined ? {BinaryValue: Buffer.from(value.binaryValue, 'base64')} : {}),
+        }]))
+        const response = await this.sqs.send(new SendMessageCommand({QueueUrl: url, MessageBody: input.messageBody, MessageAttributes: messageAttributes, MessageGroupId: input.messageGroupId, MessageDeduplicationId: input.messageDeduplicationId}))
+        return {messageId: response.MessageId ?? ''}
+    }
+
+    async receiveSqsMessages(queueId: string, input: SqsReceiveMessagesInput): Promise<SqsMessage[]> {
+        const url = await this.requireQueueUrl(queueId)
+        const max = input.maxNumberOfMessages ?? 1
+        if (!Number.isInteger(max) || max < 1 || max > 10) throw new ValidationError('maxNumberOfMessages must be between 1 and 10.')
+        const response = await this.sqs.send(new ReceiveMessageCommand({QueueUrl: url, MaxNumberOfMessages: max, MessageSystemAttributeNames: ['All'], MessageAttributeNames: ['All']}))
+        return (response.Messages ?? []).map((message) => ({
+            messageId: message.MessageId ?? '', receiptHandle: message.ReceiptHandle ?? '', body: message.Body ?? '',
+            attributes: message.Attributes ?? {},
+            messageAttributes: Object.fromEntries(Object.entries(message.MessageAttributes ?? {}).map(([name, value]) => [name, {dataType: value.DataType ?? '', stringValue: value.StringValue, binaryValue: value.BinaryValue ? Buffer.from(value.BinaryValue).toString('base64') : undefined}])),
+        }))
+    }
+
+    async deleteSqsMessage(queueId: string, receiptHandle: string): Promise<void> {
+        await this.sqs.send(new DeleteMessageCommand({QueueUrl: await this.requireQueueUrl(queueId), ReceiptHandle: receiptHandle}))
+    }
+
+    async purgeSqsQueue(queueId: string): Promise<void> {
+        await this.sqs.send(new PurgeQueueCommand({QueueUrl: await this.requireQueueUrl(queueId)}))
+    }
+
+    async startSqsMoveTask(queueId: string, input: SqsMoveTaskInput): Promise<SqsMoveTaskResult> {
+        const queue = await this.get(queueId)
+        const sourceArn = typeof queue?.metadata.arn === 'string' ? queue.metadata.arn : undefined
+        if (!sourceArn) throw new NotFoundError(`Queue ${queueId} does not exist`)
+        const response = await this.sqs.send(new StartMessageMoveTaskCommand({SourceArn: sourceArn, DestinationArn: input.destinationArn}))
+        return {taskHandle: response.TaskHandle ?? ''}
+    }
+
     /** Listing queue names is cheaper than describing each one. */
     async health(): Promise<void> {
         await this.sqs.send(new ListQueuesCommand({}))
@@ -126,6 +202,12 @@ export class AwsSqsAdapter implements CloudServiceAdapter {
         }
     }
 
+    private async requireQueueUrl(name: string): Promise<string> {
+        const url = await this.queueUrl(name)
+        if (!url) throw new NotFoundError(`Queue ${name} does not exist`)
+        return url
+    }
+
     private async attributesFor(url: string): Promise<QueueAttributes> {
         if (!url) return {}
         const res = await this.sqs.send(
@@ -157,10 +239,21 @@ function toResource(name: string, url: string, attributes: QueueAttributes): Clo
             maximumMessageSize: numberValue(attributes.MaximumMessageSize),
             delaySeconds: numberValue(attributes.DelaySeconds),
             lastModified: epochSecondsToIso(attributes.LastModifiedTimestamp),
+            redrivePolicy: parseRedrivePolicy(attributes.RedrivePolicy),
         },
     }
 }
 
+function parseRedrivePolicy(value: string | undefined): {deadLetterTargetArn?: string; maxReceiveCount?: number} | null {
+    if (!value) return null
+    try {
+        const policy = JSON.parse(value) as {deadLetterTargetArn?: unknown; maxReceiveCount?: unknown}
+        return {deadLetterTargetArn: typeof policy.deadLetterTargetArn === 'string' ? policy.deadLetterTargetArn : undefined, maxReceiveCount: typeof policy.maxReceiveCount === 'string' ? Number(policy.maxReceiveCount) : undefined}
+    } catch {
+        return null
+    }
+}
+
 /** SQS reports timestamps as epoch seconds; the contract uses ISO strings. */
 function epochSecondsToIso(value: string | undefined): string | null {
     if (!value) return null