feat: add SQS + EventBridge

2026-09-24 22:34:04 -04:00
parent 9bbe80b5f2
commit 1b129e9dcf
37 changed files with 2902 additions and 66 deletions
@@ -1,31 +1,55 @@
 import {ValidationError} from '../cloud-spi/errors'
 import {
  CreateFunctionCommand,
+  CreateEventSourceMappingCommand,
+  AddPermissionCommand,
  DeleteFunctionCommand,
+  DeleteEventSourceMappingCommand,
  GetFunctionCommand,
  InvokeCommand,
  type InvokeCommandOutput,
  ListFunctionsCommand,
+  ListEventSourceMappingsCommand,
+  RemovePermissionCommand,
+  UpdateFunctionCodeCommand,
+  UpdateFunctionConfigurationCommand,
  type LambdaClient,
 } from "@aws-sdk/client-lambda";
+import {
+  DeleteRuleCommand,
+  ListRulesCommand,
+  ListTargetsByRuleCommand,
+  PutRuleCommand,
+  PutTargetsCommand,
+  RemoveTargetsCommand,
+  type EventBridgeClient,
+} from '@aws-sdk/client-eventbridge'
 import { awsServerlessSchema } from "../cloud-spi/serverlessSchema";
 import type {
  CloudResource,
  CloudServiceAdapter,
  CreateResourceInput,
  ResourceQuery,
+  EventBridgeLambdaTriggerInput,
+  EventBridgeLambdaTriggerOverview,
+  SqsLambdaEventSourceMappingInput,
+  SqsLambdaEventSourceMappingOverview,
  ServerlessInvokeResult,
  ServiceSchema,
+  UpdateResourceInput,
 } from "../cloud-spi/types";
-import { lambda as defaultLambda } from "../aws";
-import { createZipArchive, handlerFileName } from "./zipArchive";
+import { eventbridge as defaultEventBridge, lambda as defaultLambda } from "../aws";
+import { createZipArchive, handlerFileName, readStoredZipEntry } from "./zipArchive";
 import {lambdaStarterCode} from '../cloud-spi/lambdaStarterCode'

 export class AwsServerlessAdapter implements CloudServiceAdapter {
  readonly cloud = "aws" as const;
  readonly service = "serverless" as const;

-  constructor(private readonly lambda: LambdaClient = defaultLambda) {}
+  constructor(
+    private readonly lambda: LambdaClient = defaultLambda,
+    private readonly eventBridge: EventBridgeClient = defaultEventBridge,
+  ) {}

  schema(): ServiceSchema {
    return awsServerlessSchema();
@@ -66,6 +90,7 @@ export class AwsServerlessAdapter implements CloudServiceAdapter {
      const config = res.Configuration;

      if (!config) return null;
+      const source = await deployedHandlerSource(res.Code?.Location, config.Handler, config.Runtime)

      return {
        id: config.FunctionName ?? config.FunctionArn ?? id,
@@ -89,7 +114,8 @@ export class AwsServerlessAdapter implements CloudServiceAdapter {
  role: config.Role,
  version: config.Version,
  codeSha256: config.CodeSha256,
-  environment: config.Environment?.Variables,
+   environment: config.Environment?.Variables,
+   ...(source === null ? {} : {code: source}),
 },
      };
    } catch (error) {
@@ -176,6 +202,43 @@ export class AwsServerlessAdapter implements CloudServiceAdapter {
    await this.lambda.send(new DeleteFunctionCommand({ FunctionName: id }));
  }

+  async update(id: string, input: UpdateResourceInput): Promise<CloudResource> {
+    const values = input.values;
+    const source = stringValue(values.code);
+    const zipFile = stringValue(values.zipFile);
+
+    if (source || zipFile) {
+      const current = await this.lambda.send(new GetFunctionCommand({FunctionName: id}));
+      const handler = stringValue(values.handler) || current.Configuration?.Handler;
+      const runtime = stringValue(values.runtime) || current.Configuration?.Runtime;
+      if (!handler || !runtime) throw new ValidationError('Lambda handler and runtime are required to package source code');
+
+      await this.lambda.send(new UpdateFunctionCodeCommand({
+        FunctionName: id,
+        ZipFile: zipFile
+          ? decodeBase64Zip(zipFile)
+          : createZipArchive([{name: handlerFileName(handler, runtime), content: source}]),
+      }));
+    }
+
+    const configuration = {
+      FunctionName: id,
+      ...(values.runtime === undefined ? {} : {Runtime: stringValue(values.runtime) as never}),
+      ...(values.handler === undefined ? {} : {Handler: stringValue(values.handler)}),
+      ...(values.description === undefined ? {} : {Description: stringValue(values.description)}),
+      ...(values.environment === undefined ? {} : {Environment: {Variables: environmentVariables(values.environment)}}),
+      ...(values.memorySize === undefined ? {} : {MemorySize: positiveNumber(values.memorySize, 'memorySize')}),
+      ...(values.timeout === undefined ? {} : {Timeout: positiveNumber(values.timeout, 'timeout')}),
+    };
+    if (Object.keys(configuration).length > 1) {
+      await this.lambda.send(new UpdateFunctionConfigurationCommand(configuration));
+    }
+
+    const updated = await this.get(id);
+    if (!updated) throw new ValidationError(`Lambda function ${id} was not found after update`);
+    return updated;
+  }
+
  async invoke(id: string, payload: string): Promise<ServerlessInvokeResult> {
    const startedAt = performance.now();
    const res = await this.lambda.send(
@@ -196,6 +259,157 @@ export class AwsServerlessAdapter implements CloudServiceAdapter {
      executionDuration,
    };
  }
+
+  async createEventBridgeLambdaTrigger(input: EventBridgeLambdaTriggerInput): Promise<EventBridgeLambdaTriggerOverview> {
+    const ruleName = stringValue(input.ruleName)
+    const scheduleExpression = stringValue(input.scheduleExpression)
+    const functionName = stringValue(input.functionName)
+    if (!ruleName) throw new ValidationError('ruleName is required')
+    if (!scheduleExpression) throw new ValidationError('scheduleExpression is required')
+    if (!functionName) throw new ValidationError('functionName is required')
+
+    const functionResult = await this.lambda.send(new GetFunctionCommand({FunctionName: functionName}))
+    const functionArn = functionResult.Configuration?.FunctionArn
+    if (!functionArn) throw new ValidationError(`Lambda function ${functionName} did not return an ARN`)
+
+    const rule = await this.eventBridge.send(new PutRuleCommand({
+      Name: ruleName,
+      ScheduleExpression: scheduleExpression,
+      State: 'ENABLED',
+    }))
+    if (!rule.RuleArn) throw new ValidationError(`EventBridge rule ${ruleName} did not return an ARN`)
+
+    const targetId = eventBridgeTargetId(functionName)
+    await this.eventBridge.send(new PutTargetsCommand({
+      Rule: ruleName,
+      Targets: [{Id: targetId, Arn: functionArn}],
+    }))
+    await this.ensureEventBridgePermission(functionName, ruleName, rule.RuleArn)
+    return {
+      id: ruleName,
+      ruleName,
+      ruleArn: rule.RuleArn,
+      functionName,
+      functionArn,
+      scheduleExpression,
+      state: 'ENABLED',
+    }
+  }
+
+  async listEventBridgeLambdaTriggers(functionName: string): Promise<EventBridgeLambdaTriggerOverview[]> {
+    const triggers: EventBridgeLambdaTriggerOverview[] = []
+    let nextToken: string | undefined
+    do {
+      const rules = await this.eventBridge.send(new ListRulesCommand({NextToken: nextToken}))
+      for (const rule of rules.Rules ?? []) {
+        if (!rule.Name || !rule.Arn || !rule.ScheduleExpression) continue
+        const targets = await this.listEventBridgeTargets(rule.Name)
+        for (const target of targets) {
+          if (!target.Arn || lambdaFunctionName(target.Arn) !== functionName) continue
+          triggers.push({
+            id: rule.Name,
+            ruleName: rule.Name,
+            ruleArn: rule.Arn,
+            functionName,
+            functionArn: target.Arn,
+            scheduleExpression: rule.ScheduleExpression,
+            state: rule.State ?? null,
+          })
+        }
+      }
+      nextToken = rules.NextToken
+    } while (nextToken)
+    return triggers.sort((left, right) => left.ruleName.localeCompare(right.ruleName))
+  }
+
+  async deleteEventBridgeLambdaTrigger(functionName: string, triggerId: string): Promise<void> {
+    const ruleName = stringValue(triggerId)
+    if (!ruleName) throw new ValidationError('triggerId is required')
+    const targets = await this.listEventBridgeTargets(ruleName)
+    const matchingTargets = targets.filter((target) => target.Arn && lambdaFunctionName(target.Arn) === functionName)
+    const targetIds = matchingTargets.map((target) => target.Id).filter((id): id is string => Boolean(id))
+    if (targetIds.length > 0) {
+      await this.eventBridge.send(new RemoveTargetsCommand({Rule: ruleName, Ids: targetIds}))
+    }
+    try {
+      await this.lambda.send(new RemovePermissionCommand({
+        FunctionName: functionName,
+        StatementId: eventBridgePermissionStatementId(ruleName),
+      }))
+    } catch (error) {
+      if (!hasHttpStatus(error, 404)) throw error
+    }
+    if (targets.length === matchingTargets.length) {
+      await this.eventBridge.send(new DeleteRuleCommand({Name: ruleName}))
+    }
+  }
+
+  async createSqsLambdaEventSourceMapping(input: SqsLambdaEventSourceMappingInput): Promise<SqsLambdaEventSourceMappingOverview> {
+    const functionName = stringValue(input.functionName)
+    const queueArn = stringValue(input.queueArn)
+    if (!functionName) throw new ValidationError('functionName is required')
+    if (!queueArn) throw new ValidationError('queueArn is required')
+    const batchSize = input.batchSize === undefined ? undefined : sqsBatchSize(input.batchSize)
+    const mapping = await this.lambda.send(new CreateEventSourceMappingCommand({
+      FunctionName: functionName,
+      EventSourceArn: queueArn,
+      ...(batchSize === undefined ? {} : {BatchSize: batchSize}),
+    }))
+    return sqsEventSourceMappingOverview(mapping, functionName)
+  }
+
+  async listSqsLambdaEventSourceMappings(functionName: string): Promise<SqsLambdaEventSourceMappingOverview[]> {
+    const normalizedFunctionName = stringValue(functionName)
+    if (!normalizedFunctionName) throw new ValidationError('functionName is required')
+    const mappings: SqsLambdaEventSourceMappingOverview[] = []
+    let marker: string | undefined
+    do {
+      const result = await this.lambda.send(new ListEventSourceMappingsCommand({
+        FunctionName: normalizedFunctionName,
+        Marker: marker,
+      }))
+      for (const mapping of result.EventSourceMappings ?? []) {
+        if (mapping.EventSourceArn?.includes(':sqs:')) {
+          mappings.push(sqsEventSourceMappingOverview(mapping, normalizedFunctionName))
+        }
+      }
+      marker = result.NextMarker
+    } while (marker)
+    return mappings.sort((left, right) => left.uuid.localeCompare(right.uuid))
+  }
+
+  async deleteSqsLambdaEventSourceMapping(functionName: string, mappingId: string): Promise<void> {
+    if (!stringValue(functionName)) throw new ValidationError('functionName is required')
+    const uuid = stringValue(mappingId)
+    if (!uuid) throw new ValidationError('mappingId is required')
+    await this.lambda.send(new DeleteEventSourceMappingCommand({UUID: uuid}))
+  }
+
+  private async listEventBridgeTargets(ruleName: string) {
+    const targets = []
+    let nextToken: string | undefined
+    do {
+      const result = await this.eventBridge.send(new ListTargetsByRuleCommand({Rule: ruleName, NextToken: nextToken}))
+      targets.push(...(result.Targets ?? []))
+      nextToken = result.NextToken
+    } while (nextToken)
+    return targets
+  }
+
+  private async ensureEventBridgePermission(functionName: string, ruleName: string, ruleArn: string): Promise<void> {
+    const permission = {
+      FunctionName: functionName,
+      StatementId: eventBridgePermissionStatementId(ruleName),
+      Action: 'lambda:InvokeFunction',
+      Principal: 'events.amazonaws.com',
+      SourceArn: ruleArn,
+    }
+    try {
+      await this.lambda.send(new AddPermissionCommand(permission))
+    } catch (error) {
+      if (!hasHttpStatus(error, 409)) throw error
+    }
+  }
 }

 function decodeBase64Zip(value: string): Uint8Array {
@@ -213,6 +427,99 @@ function positiveNumberOrDefault(value: unknown, defaultValue: number): number {
  return Number.isFinite(number) && number > 0 ? number : defaultValue
 }

+function positiveNumber(value: unknown, field: string): number {
+  const number = Number(value)
+  if (!Number.isFinite(number) || number <= 0) throw new ValidationError(`${field} must be a positive number`)
+  return number
+}
+
+function stringValue(value: unknown): string {
+  return typeof value === 'string' ? value.trim() : ''
+}
+
+function lambdaFunctionName(arn: string): string | null {
+  const match = arn.match(/:function:([^:/]+)/)
+  return match?.[1] ?? null
+}
+
+function eventBridgeTargetId(functionName: string): string {
+  return `floci-ui-eventbridge-${functionName}`.replace(/[^A-Za-z0-9_.-]/g, '-')
+}
+
+function eventBridgePermissionStatementId(ruleName: string): string {
+  return `floci-ui-eventbridge-${ruleName}`.replace(/[^A-Za-z0-9_.-]/g, '-').slice(0, 100)
+}
+
+function sqsBatchSize(value: number): number {
+  if (!Number.isInteger(value) || value < 1 || value > 10) {
+    throw new ValidationError('batchSize must be an integer between 1 and 10')
+  }
+  return value
+}
+
+function sqsEventSourceMappingOverview(
+  mapping: {UUID?: string; FunctionArn?: string; EventSourceArn?: string; BatchSize?: number; State?: string; StateTransitionReason?: string},
+  functionName: string,
+): SqsLambdaEventSourceMappingOverview {
+  const uuid = mapping.UUID
+  const queueArn = mapping.EventSourceArn
+  if (!uuid || !queueArn) throw new ValidationError('Lambda event source mapping did not return a UUID and SQS queue ARN')
+  return {
+    id: uuid,
+    uuid,
+    functionName,
+    functionArn: mapping.FunctionArn ?? null,
+    queueArn,
+    batchSize: mapping.BatchSize ?? null,
+    state: mapping.State ?? null,
+    stateTransitionReason: mapping.StateTransitionReason ?? null,
+  }
+}
+
+async function deployedHandlerSource(location: string | undefined, handler: string | undefined, runtime: string | undefined): Promise<string | null> {
+  if (!location || !handler || !runtime) return null
+  try {
+    const response = await globalThis.fetch(location)
+    if (!response.ok) return null
+    return readStoredZipEntry(new Uint8Array(await response.arrayBuffer()), handlerFileName(handler, runtime))
+  } catch {
+    return null
+  }
+}
+
+function environmentVariables(value: unknown): Record<string, string> {
+  let variables: unknown = value
+  if (typeof value === 'string') {
+    try {
+      variables = JSON.parse(value)
+    } catch {
+      throw new ValidationError('environment must be a JSON object of string values')
+    }
+  }
+  if (!variables || typeof variables !== 'object' || Array.isArray(variables)) {
+    if (!Array.isArray(variables)) throw new ValidationError('environment must be a JSON object of string values')
+    const entries = variables.map((entry) => {
+      if (!entry || typeof entry !== 'object' || Array.isArray(entry)) {
+        throw new ValidationError('environment rows must have non-empty keys and string values')
+      }
+      const row = entry as {key?: unknown; value?: unknown}
+      if (typeof row.key !== 'string' || !row.key || typeof row.value !== 'string') {
+        throw new ValidationError('environment rows must have non-empty keys and string values')
+      }
+      return [row.key, row.value] as const
+    })
+    if (new Set(entries.map(([key]) => key)).size !== entries.length) {
+      throw new ValidationError('environment variable names must be unique')
+    }
+    return Object.fromEntries(entries)
+  }
+  const entries = Object.entries(variables)
+  if (entries.some(([key, variable]) => !key || typeof variable !== 'string')) {
+    throw new ValidationError('environment must be a JSON object of non-empty keys and string values')
+  }
+  return Object.fromEntries(entries) as Record<string, string>
+}
+
 function decodePayload(payload: InvokeCommandOutput["Payload"]): string {
  if (!payload) return "";
  return new TextDecoder().decode(payload);