feat: implement localStack API Gateway handler for /si_events/ paths

.github/workflows/node.yml

@@ -29,6 +29,7 @@ jobs:
       - run: npm ci
       - run: npm run lint:check
       - run: npm audit --audit-level=critical
+      - run: npm run build
       - run: npm run test:ci
       - name: Upload coverage reports to Codecov
         uses: codecov/codecov-action@v4

src/commands/local.ts

@@ -1,5 +1,6 @@
-import { logger, setContext } from '../common';
+import { getIacLocation, logger, setContext } from '../common';
 import { startLocalStack } from '../stack/localStack';
+import { parseYaml } from '../parser';
 
 export type RunLocalOptions = {
   stage: string;
@@ -13,12 +14,13 @@ export const runLocal = async (stackName: string, opts: RunLocalOptions) => {
   const { stage, port, debug, watch, location } = opts;
 
   await setContext({ stage, location });
+  const iac = parseYaml(getIacLocation(location));
 
   logger.info(
     `run-local starting: stack=${stackName} stage=${stage} port=${port} debug=${debug} watch=${watch}`,
   );
 
-  await startLocalStack();
+  await startLocalStack(iac);
 
   // if (watch) {
   //   const cwd = process.cwd();

src/common/constants.ts

@@ -2,4 +2,4 @@ export const CODE_ZIP_SIZE_LIMIT = 300 * 1000; // 300 KB ROS TemplateBody size l
 export const OSS_DEPLOYMENT_TIMEOUT = 3000; // in seconds
 export const SI_BOOTSTRAP_FC_PREFIX = 'si-bootstrap-api';
 export const SI_BOOTSTRAP_BUCKET_PREFIX = 'si-bootstrap-artifacts';
-export const SI_LOCALSTACK_GATEWAY_PORT = 4567;
+export const SI_LOCALSTACK_SERVER_PORT = 4567;

src/common/iacHelper.ts

@@ -1,10 +1,11 @@
 import path from 'node:path';
 import fs from 'node:fs';
 import * as ros from '@alicloud/ros-cdk-core';
-import { Context } from '../types';
+import { Context, FunctionDomain, ServerlessIac } from '../types';
 import * as ossDeployment from '@alicloud/ros-cdk-ossdeployment';
 import crypto from 'node:crypto';
 import { get } from 'lodash';
+import { parseYaml } from '../parser';
 
 export const resolveCode = (location: string): string => {
   const filePath = path.resolve(process.cwd(), location);
@@ -96,7 +97,18 @@ export const calcValue = <T>(rawValue: string, ctx: Context): T => {
   }
 
   if (containsVar?.length) {
-    value = value.replace(/\$\{vars\.(\w+)}/g, (_, key) => getParam(key, ctx.parameters));
+    const { vars: iacVars } = parseYaml(ctx.iacLocation);
+
+    const mergedParams = Array.from(
+      new Map<string, string>(
+        [
+          ...Object.entries(iacVars ?? {}).map(([key, value]) => [key, value]),
+          ...(ctx.parameters ?? []).map(({ key, value }) => [key, value]),
+        ].filter(([, v]) => v !== undefined) as Array<[string, string]>,
+      ).entries(),
+    ).map(([key, value]) => ({ key, value }));
+
+    value = value.replace(/\$\{vars\.(\w+)}/g, (_, key) => getParam(key, mergedParams));
   }
 
   if (containsMap?.length) {
@@ -107,6 +119,17 @@ export const calcValue = <T>(rawValue: string, ctx: Context): T => {
 
   return value as T;
 };
+
+export const getIacDefinition = (
+  iac: ServerlessIac,
+  rawValue: string,
+): FunctionDomain | undefined => {
+  const matchFn = rawValue.match(/^\$\{functions\.(\w+(\.\w+)?)}$/);
+  if (matchFn?.length) {
+    return iac.functions?.find((fc) => fc.key === matchFn[1]);
+  }
+};
+
 export const formatRosId = (id: string): string => {
   // Insert underscore before uppercase letters, but only when they follow a lowercase letter
   let result = id.replace(/([a-z])([A-Z])/g, '$1_$2');
@@ -125,3 +148,11 @@ export const formatRosId = (id: string): string => {
 
   return result;
 };
+
+export const splitDomain = (domain: string) => {
+  const parts = domain.split('.');
+  const rr = parts.length > 2 ? parts[0] : '@';
+  const domainName = parts.length > 2 ? parts.slice(1).join('.') : domain;
+
+  return { rr, domainName };
+};

src/common/index.ts

@@ -8,4 +8,4 @@ export * from './constants';
 export * from './imsClient';
 export * from './base64';
 export * from './rosAssets';
-export * from './domainHelper';
+export * from './requestHelper';

src/common/requestHelper.ts

@@ -0,0 +1,14 @@
+import { IncomingMessage } from 'http';
+
+export const readRequestBody = (req: IncomingMessage): Promise<string> => {
+  return new Promise((resolve, reject) => {
+    let body = '';
+    req.on('data', (chunk) => {
+      body += chunk.toString();
+    });
+    req.on('end', () => {
+      resolve(body);
+    });
+    req.on('error', reject);
+  });
+};

src/parser/eventParser.ts

@@ -8,7 +8,7 @@ export const parseEvent = (events: { [key: string]: EventRaw }): Array<EventDoma
     key,
     name: event.name,
     type: event.type,
-    triggers: event.triggers,
+    triggers: event.triggers?.map((trigger) => ({ ...trigger, method: trigger.method ?? 'GET' })),
     domain: event.domain,
   }));
 };

src/stack/localStack/event.ts

@@ -1,38 +1,85 @@
-import http from 'node:http';
-import { logger } from '../../common';
-import { EventDomain, EventTypes } from '../../types';
+import { EventTypes, ServerlessIac } from '../../types';
 import { isEmpty } from 'lodash';
+import { ParsedRequest, RouteHandler, RouteResponse } from '../../types/localStack';
+import { IncomingMessage } from 'http';
+import { getIacDefinition, logger } from '../../common';
+import { functionsHandler } from './function';
 
-const startApiGatewayServer = (event: EventDomain) => {
-  const server = http.createServer((req, res) => {
-    const matchedTrigger = event.triggers.find(
-      (trigger) => trigger.method === req.method && trigger.path === req.url,
-    );
-    if (!matchedTrigger) {
-      res.writeHead(404, { 'Content-Type': 'text/plain; charset=utf-8' });
-      res.end('Not Found\n');
-      logger.warn(`API Gateway Event - ${req.method} ${req.url} -> Not Found`);
-      return;
-    }
+const matchTrigger = (
+  req: { method: string; path: string },
+  trigger: { method: string; path: string },
+): boolean => {
+  if (req.method !== 'ANY' && req.method !== trigger.method) {
+    return false;
+  }
 
-    res.writeHead(200, { 'Content-Type': 'text/plain; charset=utf-8' });
-    res.end(`Invoked backend: ${matchedTrigger.backend}\n`);
-    logger.info(`API Gateway Event - ${req.method} ${req.url} -> ${matchedTrigger.backend}`);
-  });
+  const normalize = (s: string) => s.replace(/^\/+|\/+$/g, '');
+  const [pathSegments, triggerSegments] = [
+    normalize(req.path).split('/'),
+    normalize(trigger.path).split('/'),
+  ];
+
+  const hasWildcard = triggerSegments[triggerSegments.length - 1] === '*';
+
+  const prefixSegments = hasWildcard ? triggerSegments.slice(0, -1) : triggerSegments;
+  const minRequiredSegments = prefixSegments.length;
+
+  if (pathSegments.length < minRequiredSegments) return false;
+
+  return prefixSegments.every((triggerSegment, index) => {
+    const pathSegment = pathSegments[index];
+
+    if (triggerSegment.startsWith('[') && triggerSegment.endsWith(']')) {
+      return pathSegment !== '';
+    }
 
-  const port = 3000 + Math.floor(Math.random() * 1000);
-  server.listen(port, () => {
-    logger.info(`API Gateway "${event.name}" listening on http://localhost:${port}`);
+    return triggerSegment === pathSegment;
   });
 };
 
-export const startEvents = (events: Array<EventDomain> | undefined) => {
-  const apiGateways = events?.filter((event) => event.type === EventTypes.API_GATEWAY);
-  if (isEmpty(apiGateways)) {
-    return;
+const servEvent = async (
+  req: IncomingMessage,
+  parsed: ParsedRequest,
+  iac: ServerlessIac,
+): Promise<RouteResponse | void> => {
+  const event = iac.events?.find(
+    (event) => event.type === EventTypes.API_GATEWAY && event.key === parsed.identifier,
+  );
+
+  if (isEmpty(event)) {
+    return {
+      statusCode: 404,
+      body: { error: 'API Gateway event not found', event: parsed.identifier },
+    };
   }
+  logger.info(
+    `Event trigger ${JSON.stringify(event.triggers)}, req method: ${req.method}, req url${req.url}`,
+  );
+  const matchedTrigger = event.triggers.find((trigger) =>
+    matchTrigger({ method: parsed.method, path: parsed.url }, trigger),
+  );
 
-  apiGateways!.forEach((gateway) => {
-    startApiGatewayServer(gateway);
-  });
+  if (!matchedTrigger) {
+    return { statusCode: 404, body: { error: 'No matching trigger found' } };
+  }
+
+  if (matchedTrigger.backend) {
+    const backendDef = getIacDefinition(iac, matchedTrigger.backend);
+    if (!backendDef) {
+      return {
+        statusCode: 500,
+        body: { error: 'Backend definition missing', backend: matchedTrigger.backend },
+      };
+    }
+    return await functionsHandler(req, { ...parsed, identifier: backendDef?.key as string }, iac);
+  }
+
+  return {
+    statusCode: 202,
+    body: { message: 'Trigger matched but no backend configured' },
+  };
+};
+
+export const eventsHandler: RouteHandler = async (req, parsed, iac) => {
+  return await servEvent(req, parsed, iac);
 };

src/stack/localStack/function.ts

@@ -0,0 +1,141 @@
+import { IncomingMessage } from 'http';
+import { ServerlessIac } from '../../types';
+import { FunctionOptions, ParsedRequest, RouteResponse } from '../../types/localStack';
+import { logger, getContext, calcValue, readRequestBody } from '../../common';
+import { invokeFunction } from './functionRunner';
+import path from 'node:path';
+import fs from 'node:fs';
+import JSZip from 'jszip';
+import os from 'node:os';
+
+const extractZipFile = async (zipPath: string): Promise<string> => {
+  const zipData = fs.readFileSync(zipPath);
+  const zip = await JSZip.loadAsync(zipData);
+
+  const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'si-function-'));
+
+  for (const [relativePath, file] of Object.entries(zip.files)) {
+    if (file.dir) {
+      fs.mkdirSync(path.join(tempDir, relativePath), { recursive: true });
+    } else {
+      const content = await file.async('nodebuffer');
+      const filePath = path.join(tempDir, relativePath);
+      fs.mkdirSync(path.dirname(filePath), { recursive: true });
+      fs.writeFileSync(filePath, content);
+    }
+  }
+
+  // Check if there's a single root directory in the zip
+  // If so, return that directory instead of the temp directory
+  const entries = fs.readdirSync(tempDir);
+  if (entries.length === 1) {
+    const singleEntry = path.join(tempDir, entries[0]);
+    if (fs.statSync(singleEntry).isDirectory()) {
+      return singleEntry;
+    }
+  }
+
+  return tempDir;
+};
+
+export const functionsHandler = async (
+  req: IncomingMessage,
+  parsed: ParsedRequest,
+  iac: ServerlessIac,
+): Promise<RouteResponse> => {
+  logger.info(
+    `Function request received by local server -> ${req.method} ${parsed.identifier ?? '/'} `,
+  );
+
+  const fcDef = iac.functions?.find((fn) => fn.key === parsed.identifier);
+  if (!fcDef) {
+    return {
+      statusCode: 404,
+      body: { error: 'Function not found', functionKey: parsed.identifier },
+    };
+  }
+
+  if (!fcDef.code) {
+    return {
+      statusCode: 400,
+      body: { error: 'Function code configuration not found', functionKey: fcDef.key },
+    };
+  }
+
+  let tempDir: string | null = null;
+
+  try {
+    const rawBody = await readRequestBody(req);
+    const event = rawBody ? JSON.parse(rawBody) : {};
+
+    const ctx = getContext();
+    logger.debug(`Context parameters: ${JSON.stringify(ctx.parameters)}`);
+
+    const codePath = path.resolve(process.cwd(), calcValue(fcDef.code.path, ctx));
+
+    let codeDir: string;
+
+    if (codePath.endsWith('.zip') && fs.existsSync(codePath)) {
+      tempDir = await extractZipFile(codePath);
+      codeDir = tempDir;
+    } else if (fs.existsSync(codePath) && fs.statSync(codePath).isDirectory()) {
+      codeDir = codePath;
+    } else {
+      codeDir = path.dirname(codePath);
+    }
+    const functionName = calcValue<string>(fcDef.name, ctx);
+
+    const funOptions: FunctionOptions = {
+      codeDir,
+      functionKey: fcDef.key,
+      handler: calcValue(fcDef.code.handler, ctx),
+      servicePath: '',
+      timeout: (fcDef.timeout || 3) * 1000,
+    };
+
+    const env = {
+      ...fcDef.environment,
+      AWS_REGION: iac.provider.region || 'us-east-1',
+      FUNCTION_NAME: functionName,
+      FUNCTION_MEMORY_SIZE: String(fcDef.memory || 128),
+      FUNCTION_TIMEOUT: String(fcDef.timeout || 3),
+    };
+
+    const fcContext = {
+      functionName,
+      functionVersion: '$LATEST',
+      memoryLimitInMB: fcDef.memory || 128,
+      logGroupName: `/aws/lambda/${functionName}`,
+      logStreamName: `${new Date().toISOString().split('T')[0]}/[$LATEST]${Math.random().toString(36).substring(7)}`,
+      invokedFunctionArn: `arn:aws:lambda:${iac.provider.region}:000000000000:function:${functionName}`,
+      awsRequestId: Math.random().toString(36).substring(2, 15),
+    };
+
+    logger.debug(
+      `Invoking worker with event: ${JSON.stringify(event)} and context: ${JSON.stringify(fcContext)}`,
+    );
+    logger.debug(`Worker codeDir: ${codeDir}, handler: ${funOptions.handler}`);
+
+    const result = await invokeFunction(funOptions, env, event, fcContext);
+
+    logger.info(`Function execution result: ${JSON.stringify(result)}`);
+
+    return {
+      statusCode: 200,
+      body: result,
+    };
+  } catch (error) {
+    logger.error(`Function execution error: ${error}`);
+    return {
+      statusCode: 500,
+      body: {
+        error: 'Function execution failed',
+        message: error instanceof Error ? error.message : String(error),
+      },
+    };
+  } finally {
+    if (tempDir && fs.existsSync(tempDir)) {
+      fs.rmSync(tempDir, { recursive: true, force: true });
+    }
+  }
+};