Feature: sendEvent can use an AWS SQS queue to increase reliability and throughput (#729)

* Initial commit (by create-cloudflare CLI)

* Changed the prettier rules

* Started writing Readme and got dev working

* Validate the request, parse the event and respond with the correct format

* Use ulidx instead of ulid in core

* Removed some boilerplate and use new function for sendEvent

* Removed old ulid package

* When handling a webhook, it’s not an error if the HTTP endpoint environment isn’t found

* If the env vars aren’t set, log the request and pass it through

* Pass requests through. Setup the env vars

* Flipped the API key detection because we can enforce that it’s the shape of a private API key

* Progress with sendEvent proxy

* Events are being put on the SQS queue

* The SQS event queue is working

* Make sure there’s a timestamp before enqueuing the evnet

* Log the sqs_event, not API key

* Added a bit more detail to the readme

* Improved the Cloudflare logs

* We don’t need the global.window hack for AWS, just globalThis

* Removed commented out wrangler.toml values

* Reworked the proxy to make it easier to add more endpoints

* Use json utility to send Responses

* Start work on proxying bulk events

apps/proxy/.dev.vars.example

@@ -0,0 +1,7 @@
+REWRITE_HOSTNAME=
+AWS_SQS_ACCESS_KEY_ID=
+AWS_SQS_SECRET_ACCESS_KEY=
+AWS_SQS_QUEUE_URL=
+AWS_SQS_REGION=
+#optional
+#REWRITE_PORT=

apps/proxy/.editorconfig

@@ -0,0 +1,13 @@
+# http://editorconfig.org
+root = true
+
+[*]
+indent_style = tab
+tab_width = 2
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
+
+[*.yml]
+indent_style = space

apps/proxy/.gitignore

@@ -0,0 +1,172 @@
+# Logs
+
+logs
+_.log
+npm-debug.log_
+yarn-debug.log*
+yarn-error.log*
+lerna-debug.log*
+.pnpm-debug.log*
+
+# Diagnostic reports (https://nodejs.org/api/report.html)
+
+report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json
+
+# Runtime data
+
+pids
+_.pid
+_.seed
+\*.pid.lock
+
+# Directory for instrumented libs generated by jscoverage/JSCover
+
+lib-cov
+
+# Coverage directory used by tools like istanbul
+
+coverage
+\*.lcov
+
+# nyc test coverage
+
+.nyc_output
+
+# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files)
+
+.grunt
+
+# Bower dependency directory (https://bower.io/)
+
+bower_components
+
+# node-waf configuration
+
+.lock-wscript
+
+# Compiled binary addons (https://nodejs.org/api/addons.html)
+
+build/Release
+
+# Dependency directories
+
+node_modules/
+jspm_packages/
+
+# Snowpack dependency directory (https://snowpack.dev/)
+
+web_modules/
+
+# TypeScript cache
+
+\*.tsbuildinfo
+
+# Optional npm cache directory
+
+.npm
+
+# Optional eslint cache
+
+.eslintcache
+
+# Optional stylelint cache
+
+.stylelintcache
+
+# Microbundle cache
+
+.rpt2_cache/
+.rts2_cache_cjs/
+.rts2_cache_es/
+.rts2_cache_umd/
+
+# Optional REPL history
+
+.node_repl_history
+
+# Output of 'npm pack'
+
+\*.tgz
+
+# Yarn Integrity file
+
+.yarn-integrity
+
+# dotenv environment variable files
+
+.env
+.env.development.local
+.env.test.local
+.env.production.local
+.env.local
+
+# parcel-bundler cache (https://parceljs.org/)
+
+.cache
+.parcel-cache
+
+# Next.js build output
+
+.next
+out
+
+# Nuxt.js build / generate output
+
+.nuxt
+dist
+
+# Gatsby files
+
+.cache/
+
+# Comment in the public line in if your project uses Gatsby and not Next.js
+
+# https://nextjs.org/blog/next-9-1#public-directory-support
+
+# public
+
+# vuepress build output
+
+.vuepress/dist
+
+# vuepress v2.x temp and cache directory
+
+.temp
+.cache
+
+# Docusaurus cache and generated files
+
+.docusaurus
+
+# Serverless directories
+
+.serverless/
+
+# FuseBox cache
+
+.fusebox/
+
+# DynamoDB Local files
+
+.dynamodb/
+
+# TernJS port file
+
+.tern-port
+
+# Stores VSCode versions used for testing VSCode extensions
+
+.vscode-test
+
+# yarn v2
+
+.yarn/cache
+.yarn/unplugged
+.yarn/build-state.yml
+.yarn/install-state.gz
+.pnp.\*
+
+# wrangler project
+
+.dev.vars
+.wrangler/

apps/proxy/.prettierrc

@@ -0,0 +1,11 @@
+{
+	"semi": true,
+	"singleQuote": false,
+	"jsxSingleQuote": false,
+	"trailingComma": "es5",
+	"bracketSpacing": true,
+	"bracketSameLine": false,
+	"printWidth": 100,
+	"tabWidth": 2,
+	"useTabs": false
+}

apps/proxy/README.md

@@ -0,0 +1,52 @@
+# Trigger.dev proxy
+
+This is an optional module that can be used to proxy and queue requests to the Trigger.dev API.
+
+## Why?
+
+The Trigger.dev API is designed to be fast and reliable. However, if you have a lot of traffic, you may want to use this proxy to queue requests to the API. It intercepts some requests to the API and adds them to an AWS SQS queue, then the webapp can be setup to process the queue.
+
+## Current features
+
+- Intercepts `sendEvent` requests and adds them to an AWS SQS queue. The webapp then reads from the queue and creates the events.
+
+## Setup
+
+### 1. Create an AWS SQS queue
+
+In AWS you should create a new AWS SQS queue with appropriate security settings. You will need the queue URL for the next step.
+
+### Environment variables
+
+#### Cloudflare secrets
+
+Locally you should copy the `.dev.var.example` file to `.dev.var` and fill in the values.
+
+When deploying you should use `wrangler` (the Cloudflare CLI tool) to set secrets.
+
+```bash
+wrangler secret put REWRITE_HOSTNAME
+wrangler secret put AWS_SQS_ACCESS_KEY_ID
+wrangler secret put AWS_SQS_SECRET_ACCESS_KEY
+wrangler secret put AWS_SQS_QUEUE_URL
+wrangler secret put AWS_SQS_REGION
+```
+
+#### Webapp
+
+These env vars also need setting in the webapp, however you normally would do that.
+
+```bash
+AWS_SQS_REGION
+AWS_SQS_ACCESS_KEY_ID
+AWS_SQS_SECRET_ACCESS_KEY
+AWS_SQS_QUEUE_URL
+AWS_SQS_BATCH_SIZE
+```
+
+## Development
+
+Set the environment variables as described above.
+
+1. `pnpm install`
+2. `pnpm run dev --filter proxy`

apps/proxy/package.json

@@ -0,0 +1,21 @@
+{
+  "name": "proxy",
+  "version": "0.0.0",
+  "private": true,
+  "scripts": {
+    "deploy": "wrangler deploy",
+    "dev": "wrangler dev"
+  },
+  "devDependencies": {
+    "@cloudflare/workers-types": "^4.20230419.0",
+    "typescript": "^5.0.4",
+    "wrangler": "^3.0.0"
+  },
+  "dependencies": {
+    "@aws-sdk/client-sqs": "^3.445.0",
+    "@trigger.dev/core": "workspace:*",
+    "ulidx": "^2.2.1",
+    "zod": "3.22.3",
+    "zod-error": "1.5.0"
+  }
+}

apps/proxy/src/apikey.ts

@@ -0,0 +1,20 @@
+import { z } from "zod";
+
+const AuthorizationHeaderSchema = z.string().regex(/^Bearer .+$/);
+
+export function getApiKeyFromRequest(request: Request) {
+  const rawAuthorization = request.headers.get("Authorization");
+
+  const authorization = AuthorizationHeaderSchema.safeParse(rawAuthorization);
+  if (!authorization.success) {
+    return;
+  }
+
+  const apiKey = authorization.data.replace(/^Bearer /, "");
+  const type = isPrivateApiKey(apiKey) ? ("PRIVATE" as const) : ("PUBLIC" as const);
+  return { apiKey, type };
+}
+
+function isPrivateApiKey(key: string) {
+  return key.startsWith("tr_");
+}

apps/proxy/src/events/queueEvent.ts

@@ -0,0 +1,88 @@
+import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
+import { ApiEventLog, SendEventBodySchema, SendEventOptions } from "@trigger.dev/core";
+import { generateErrorMessage } from "zod-error";
+import { getApiKeyFromRequest } from "../apikey";
+import { Env } from "..";
+import { calculateDeliverAt } from "./utils";
+import { json } from "../json";
+
+/** Adds the event to an AWS SQS queue, so it can be consumed from the main Trigger.dev API */
+export async function queueEvent(request: Request, env: Env): Promise<Response> {
+  //check there's a private API key
+  const apiKeyResult = getApiKeyFromRequest(request);
+  if (!apiKeyResult || apiKeyResult.type !== "PRIVATE") {
+    return json(
+      { error: "Invalid or Missing API key" },
+      {
+        status: 401,
+      }
+    );
+  }
+
+  //parse the request body
+  try {
+    const anyBody = await request.json();
+    const body = SendEventBodySchema.safeParse(anyBody);
+    if (!body.success) {
+      return json(
+        { message: generateErrorMessage(body.error.issues) },
+        {
+          status: 422,
+        }
+      );
+    }
+
+    // The AWS SDK tries to use crypto from off of the window,
+    // so we need to trick it into finding it where it expects it
+    globalThis.global = globalThis;
+
+    const client = new SQSClient({
+      region: env.AWS_SQS_REGION,
+      credentials: {
+        accessKeyId: env.AWS_SQS_ACCESS_KEY_ID,
+        secretAccessKey: env.AWS_SQS_SECRET_ACCESS_KEY,
+      },
+    });
+
+    const timestamp = body.data.event.timestamp ?? new Date();
+
+    //add the event to the queue
+    const send = new SendMessageCommand({
+      // use wrangler secrets to provide this global variable
+      QueueUrl: env.AWS_SQS_QUEUE_URL,
+      MessageBody: JSON.stringify({
+        event: { ...body.data.event, timestamp },
+        options: body.data.options,
+        apiKey: apiKeyResult.apiKey,
+      }),
+    });
+
+    const queuedEvent = await client.send(send);
+    console.log("Queued event", queuedEvent);
+
+    //respond with the event
+    const event: ApiEventLog = {
+      id: body.data.event.id,
+      name: body.data.event.name,
+      payload: body.data.event.payload,
+      context: body.data.event.context,
+      timestamp,
+      deliverAt: calculateDeliverAt(body.data.options),
+    };
+
+    return json(event, {
+      status: 200,
+    });
+  } catch (e) {
+    return json(
+      {
+        message: `Failed to parse event body: ${
+          e instanceof Error ? e.message : JSON.stringify(e)
+        }`,
+      },
+      {
+        status: 422,
+      }
+    );
+  }
+}

apps/proxy/src/events/queueEvents.ts

@@ -0,0 +1,99 @@
+import { SQSClient, SendMessageBatchCommand } from "@aws-sdk/client-sqs";
+import {
+  ApiEventLog,
+  SendBulkEventsBodySchema,
+  SendEventBodySchema,
+  SendEventOptions,
+} from "@trigger.dev/core";
+import { generateErrorMessage } from "zod-error";
+import { getApiKeyFromRequest } from "../apikey";
+import { Env } from "..";
+import { calculateDeliverAt } from "./utils";
+import { json } from "../json";
+
+/** Adds the event to an AWS SQS queue, so it can be consumed from the main Trigger.dev API */
+export async function queueEvents(request: Request, env: Env): Promise<Response> {
+  //check there's a private API key
+  const apiKeyResult = getApiKeyFromRequest(request);
+  if (!apiKeyResult || apiKeyResult.type !== "PRIVATE") {
+    return json(
+      { error: "Invalid or Missing API key" },
+      {
+        status: 401,
+      }
+    );
+  }
+
+  //parse the request body
+  try {
+    const anyBody = await request.json();
+    const body = SendBulkEventsBodySchema.safeParse(anyBody);
+    if (!body.success) {
+      return json(
+        { message: generateErrorMessage(body.error.issues) },
+        {
+          status: 422,
+        }
+      );
+    }
+
+    // The AWS SDK tries to use crypto from off of the window,
+    // so we need to trick it into finding it where it expects it
+    globalThis.global = globalThis;
+
+    const client = new SQSClient({
+      region: env.AWS_SQS_REGION,
+      credentials: {
+        accessKeyId: env.AWS_SQS_ACCESS_KEY_ID,
+        secretAccessKey: env.AWS_SQS_SECRET_ACCESS_KEY,
+      },
+    });
+
+    const updatedEvents = body.data.events.map((event) => {
+      const timestamp = event.timestamp ?? new Date();
+      return {
+        ...event,
+        timestamp,
+      };
+    });
+
+    //add the event to the queue
+    const send = new SendMessageBatchCommand({
+      // use wrangler secrets to provide this global variable
+      QueueUrl: env.AWS_SQS_QUEUE_URL,
+      Entries: updatedEvents.map((event) => ({
+        Id: event.id,
+        MessageBody: JSON.stringify({
+          event,
+          options: body.data.options,
+          apiKey: apiKeyResult.apiKey,
+        }),
+      })),
+    });
+
+    const queuedEvent = await client.send(send);
+    console.log("Queued event", queuedEvent);
+
+    //respond with the events
+    const events: ApiEventLog[] = updatedEvents.map((event) => ({
+      ...event,
+      payload: event.payload,
+      deliverAt: calculateDeliverAt(body.data.options),
+    }));
+
+    return json(events, {
+      status: 200,
+    });
+  } catch (e) {
+    return json(
+      {
+        message: `Failed to parse event body: ${
+          e instanceof Error ? e.message : JSON.stringify(e)
+        }`,
+      },
+      {
+        status: 422,
+      }
+    );
+  }
+}

apps/proxy/src/events/utils.ts

@@ -0,0 +1,15 @@
+import { SendEventOptions } from "@trigger.dev/core";
+
+export function calculateDeliverAt(options?: SendEventOptions) {
+  // If deliverAt is a string and a valid date, convert it to a Date object
+  if (options?.deliverAt) {
+    return options?.deliverAt;
+  }
+
+  // deliverAfter is the number of seconds to wait before delivering the event
+  if (options?.deliverAfter) {
+    return new Date(Date.now() + options.deliverAfter * 1000);
+  }
+
+  return undefined;
+}

apps/proxy/src/index.ts

@@ -0,0 +1,62 @@
+import { queueEvent } from "./events/queueEvent";
+import { queueEvents } from "./events/queueEvents";
+
+export interface Env {
+  /** The hostname needs to be changed to allow requests to pass to the Trigger.dev platform */
+  REWRITE_HOSTNAME: string;
+  REWRITE_PORT?: string;
+  AWS_SQS_ACCESS_KEY_ID: string;
+  AWS_SQS_SECRET_ACCESS_KEY: string;
+  AWS_SQS_QUEUE_URL: string;
+  AWS_SQS_REGION: string;
+}
+
+export default {
+  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
+    if (!env.REWRITE_HOSTNAME) throw new Error("Missing REWRITE_HOSTNAME");
+    console.log("url", request.url);
+
+    if (!queueingIsEnabled(env)) {
+      console.log("Missing AWS credentials. Passing through to the origin.");
+      return redirectToOrigin(request, env);
+    }
+
+    const url = new URL(request.url);
+    switch (url.pathname) {
+      case "/api/v1/events": {
+        if (request.method === "POST") {
+          return queueEvent(request, env);
+        }
+        break;
+      }
+      case "/api/v1/events/bulk": {
+        if (request.method === "POST") {
+          return queueEvents(request, env);
+        }
+        break;
+      }
+    }
+
+    //the same request but with the hostname (and port) changed
+    return redirectToOrigin(request, env);
+  },
+};
+
+function redirectToOrigin(request: Request, env: Env) {
+  const newUrl = new URL(request.url);
+  newUrl.hostname = env.REWRITE_HOSTNAME;
+  newUrl.port = env.REWRITE_PORT || newUrl.port;
+
+  const requestInit: RequestInit = {
+    method: request.method,
+    headers: request.headers,
+    body: request.body,
+  };
+
+  console.log("rewritten url", newUrl.toString());
+  return fetch(newUrl.toString(), requestInit);
+}
+
+function queueingIsEnabled(env: Env) {
+  return env.AWS_SQS_ACCESS_KEY_ID && env.AWS_SQS_SECRET_ACCESS_KEY && env.AWS_SQS_QUEUE_URL;
+}

apps/proxy/src/json.ts

@@ -0,0 +1,13 @@
+export function json(body: any, init?: ResponseInit) {
+  const headers = {
+    "content-type": "application/json",
+    ...(init?.headers ?? {}),
+  };
+
+  const responseInit: ResponseInit = {
+    ...(init ?? {}),
+    headers,
+  };
+
+  return new Response(JSON.stringify(body), responseInit);
+}

apps/proxy/tsconfig.json

@@ -0,0 +1,34 @@
+{
+  "compilerOptions": {
+    "target": "es2021" /* Set the JavaScript language version for emitted JavaScript and include compatible library declarations. */,
+    "lib": [
+      "es2021"
+    ] /* Specify a set of bundled library declaration files that describe the target runtime environment. */,
+    "jsx": "react" /* Specify what JSX code is generated. */,
+
+    "module": "es2022" /* Specify what module code is generated. */,
+    "moduleResolution": "node" /* Specify how TypeScript looks up a file from a given module specifier. */,
+
+    "types": [
+      "@cloudflare/workers-types"
+    ] /* Specify type package names to be included without being referenced in a source file. */,
+    "resolveJsonModule": true /* Enable importing .json files */,
+
+    "allowJs": true /* Allow JavaScript files to be a part of your program. Use the `checkJS` option to get errors from these files. */,
+    "checkJs": false /* Enable error reporting in type-checked JavaScript files. */,
+
+    "noEmit": true /* Disable emitting files from a compilation. */,
+
+    "isolatedModules": true /* Ensure that each file can be safely transpiled without relying on other imports. */,
+    "allowSyntheticDefaultImports": true /* Allow 'import x from y' when a module doesn't have a default export. */,
+    "forceConsistentCasingInFileNames": true /* Ensure that casing is correct in imports. */,
+
+    "strict": true /* Enable all strict type-checking options. */,
+
+    "skipLibCheck": true /* Skip type checking all .d.ts files. */,
+    "paths": {
+      "@trigger.dev/core": ["../../packages/core/src/index"],
+      "@trigger.dev/core/*": ["../../packages/core/src/*"]
+    }
+  }
+}

apps/proxy/wrangler.toml

@@ -0,0 +1,3 @@
+name = "proxy"
+main = "src/index.ts"
+compatibility_date = "2023-10-30"

apps/webapp/app/entry.server.tsx

@@ -16,6 +16,8 @@ import {
   OperatingSystemPlatform,
 } from "./components/primitives/OperatingSystemProvider";
 import { env } from "./env.server";
+import { getSharedSqsEventConsumer } from "./services/events/sqsEventConsumer";
+import { singleton } from "./utils/singleton";
 
 const ABORT_DELAY = 30000;
 
@@ -190,3 +192,5 @@ function logError(error: unknown, request?: Request) {
   }
   console.error(error);
 }
+
+const sqsEventConsumer = singleton("sqsEventConsumer", getSharedSqsEventConsumer);

apps/webapp/app/env.server.ts

@@ -49,6 +49,15 @@ const EnvironmentSchema = z.object({
   TASK_OPERATION_WORKER_CONCURRENCY: z.coerce.number().int().default(10),
   TASK_OPERATION_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
   GRACEFUL_SHUTDOWN_TIMEOUT: z.coerce.number().int().default(60000),
+  /** Optional. Only used if you use the apps/proxy */
+  AWS_SQS_REGION: z.string().optional(),
+  /** Optional. Only used if you use the apps/proxy */
+  AWS_SQS_ACCESS_KEY_ID: z.string().optional(),
+  /** Optional. Only used if you use the apps/proxy */
+  AWS_SQS_SECRET_ACCESS_KEY: z.string().optional(),
+  /** Optional. Only used if you use the apps/proxy */
+  AWS_SQS_QUEUE_URL: z.string().optional(),
+  AWS_SQS_BATCH_SIZE: z.coerce.number().int().optional().default(10),
   DISABLE_SSE: z.string().optional(),
 });
 

apps/webapp/app/services/apiAuth.server.ts

@@ -20,7 +20,19 @@ export async function authenticateApiRequest(
   request: Request,
   { allowPublicKey = false }: { allowPublicKey?: boolean } = {}
 ): Promise<ApiAuthenticationResult | undefined> {
-  const result = getApiKeyFromRequest(request);
+  const apiKey = getApiKeyFromRequest(request);
+  if (!apiKey) {
+    return;
+  }
+
+  return authenticateApiKey(apiKey, { allowPublicKey });
+}
+
+export async function authenticateApiKey(
+  apiKey: string,
+  { allowPublicKey = false }: { allowPublicKey?: boolean } = {}
+): Promise<ApiAuthenticationResult | undefined> {
+  const result = getApiKeyResult(apiKey);
 
   if (!result) {
     return;
@@ -69,6 +81,10 @@ export function getApiKeyFromRequest(request: Request) {
   }
 
   const apiKey = authorization.data.replace(/^Bearer /, "");
+  return apiKey;
+}
+
+export function getApiKeyResult(apiKey: string) {
   const type = isPublicApiKey(apiKey) ? ("PUBLIC" as const) : ("PRIVATE" as const);
   return { apiKey, type };
 }

apps/webapp/app/services/events/sqsEventConsumer.ts

@@ -0,0 +1,149 @@
+import { Consumer } from "sqs-consumer";
+import { PrismaClientOrTransaction, prisma } from "~/db.server";
+import { logger } from "../logger.server";
+import { Message, SQSClient } from "@aws-sdk/client-sqs";
+import { authenticateApiKey } from "../apiAuth.server";
+import { SendEventBodySchema } from "@trigger.dev/core";
+import { z } from "zod";
+import { fromZodError } from "zod-validation-error";
+import { IngestSendEvent } from "./ingestSendEvent.server";
+import { env } from "~/env.server";
+import { singleton } from "~/utils/singleton";
+
+type SqsEventConsumerOptions = {
+  queueUrl: string;
+  /** This cannot be higher than the AWS limit of 10. */
+  batchSize: number;
+  region: string;
+  accessKeyId: string;
+  secretAccessKey: string;
+};
+
+const messageSchema = SendEventBodySchema.extend({
+  apiKey: z.string(),
+});
+
+export class SqsEventConsumer {
+  readonly #ingestEventService: IngestSendEvent;
+  readonly #consumer: Consumer;
+
+  constructor(
+    readonly prismaClient: PrismaClientOrTransaction = prisma,
+    options: SqsEventConsumerOptions
+  ) {
+    this.#ingestEventService = new IngestSendEvent();
+
+    logger.debug("SqsEventConsumer starting", {
+      queueUrl: options.queueUrl,
+      region: options.region,
+    });
+
+    this.#consumer = Consumer.create({
+      queueUrl: options.queueUrl,
+      batchSize: options.batchSize,
+      sqs: new SQSClient({
+        region: options.region,
+        credentials: {
+          accessKeyId: options.accessKeyId,
+          secretAccessKey: options.secretAccessKey,
+        },
+      }),
+      handleMessage: async (message) => {
+        await this.#processEvent(message);
+      },
+    });
+
+    this.#consumer.on("error", (err) => {
+      logger.error("SqsEventConsumer error", { message: err.message });
+      //todo what do we want to do here?
+    });
+
+    this.#consumer.on("processing_error", (err) => {
+      logger.error("SqsEventConsumer processing_error", { message: err.message });
+      //todo what do we want to do here?
+    });
+
+    this.#consumer.on("timeout_error", (err) => {
+      logger.error("SqsEventConsumer timeout_error", { message: err.message });
+      //todo what do we want to do here?
+    });
+
+    //Stop the consumer if the process is terminated
+    process.on("SIGTERM", () => {
+      this.stop();
+    });
+
+    this.#consumer.start();
+  }
+
+  public stop() {
+    logger.debug("SqsEventConsumer stopping");
+    this.#consumer.stop({ abort: true });
+  }
+
+  async #processEvent(message: Message) {
+    logger.debug("SqsEventConsumer processing event", { message });
+
+    //parse the body
+    if (!message.Body) {
+      logger.error("SqsEventConsumer message has no body", { message });
+      return;
+    }
+
+    const body = messageSchema.safeParse(JSON.parse(message.Body));
+    if (!body.success) {
+      logger.error("SqsEventConsumer message body is invalid", {
+        message,
+        error: fromZodError(body.error).message,
+      });
+      return;
+    }
+
+    //authenticate API Key
+    const authenticationResult = await authenticateApiKey(body.data.apiKey);
+    if (!authenticationResult) {
+      logger.warn("SqsEventConsumer message has invalid API key", { message });
+      return;
+    }
+
+    const authenticatedEnv = authenticationResult.environment;
+
+    logger.info("sqs_event", { event: body.data.event, options: body.data.options });
+
+    const event = await this.#ingestEventService.call(
+      authenticatedEnv,
+      body.data.event,
+      body.data.options
+    );
+
+    if (!event) {
+      logger.error("SqsEventConsumer failed to create event", { message });
+      return;
+    }
+
+    logger.debug("SqsEventConsumer processed event", { event });
+  }
+}
+
+export function getSharedSqsEventConsumer() {
+  if (
+    env.AWS_SQS_QUEUE_URL &&
+    env.AWS_SQS_REGION &&
+    env.AWS_SQS_ACCESS_KEY_ID &&
+    env.AWS_SQS_SECRET_ACCESS_KEY
+  ) {
+    const consumer = new SqsEventConsumer(undefined, {
+      queueUrl: env.AWS_SQS_QUEUE_URL,
+      batchSize: env.AWS_SQS_BATCH_SIZE,
+      region: env.AWS_SQS_REGION,
+      accessKeyId: env.AWS_SQS_ACCESS_KEY_ID,
+      secretAccessKey: env.AWS_SQS_SECRET_ACCESS_KEY,
+    });
+
+    return consumer;
+  }
+
+  console.log(
+    "The SqsEventConsumer is disabled because AWS credentials are missing. This is OK as this is an optional feature."
+  );
+}

apps/webapp/app/services/httpendpoint/HandleHttpEndpointService.ts

@@ -80,7 +80,7 @@ export class HandleHttpEndpointService {
       });
 
     if (!httpEndpointEnvironment) {
-      logger.error("Could not find http endpoint environment", {
+      logger.debug("Could not find http endpoint environment", {
         httpEndpointId: httpEndpoint.id,
         environmentId: environment.id,
       });

apps/webapp/package.json

@@ -29,6 +29,7 @@
     "/public/build"
   ],
   "dependencies": {
+    "@aws-sdk/client-sqs": "^3.445.0",
     "@codemirror/autocomplete": "^6.3.1",
     "@codemirror/commands": "^6.1.2",
     "@codemirror/lang-javascript": "^6.1.1",
@@ -111,6 +112,7 @@
     "simplur": "^3.0.1",
     "slug": "^6.0.0",
     "sonner": "^1.0.3",
+    "sqs-consumer": "^7.4.0",
     "tailwind-merge": "^1.12.0",
     "tailwind-scrollbar-hide": "^1.1.7",
     "tailwindcss-animate": "^1.0.5",

packages/core/package.json

@@ -27,7 +27,7 @@
     "test": "jest"
   },
   "dependencies": {
-    "ulid": "^2.3.0",
+    "ulidx": "^2.2.1",
     "zod": "3.22.3",
     "zod-error": "1.5.0"
   },
@@ -44,4 +44,4 @@
   "engines": {
     "node": ">=18.0.0"
   }
-}
+}

packages/core/src/schemas/api.ts

@@ -1,4 +1,4 @@
-import { ulid } from "ulid";
+import { ulid } from "ulidx";
 import { z } from "zod";
 import { Prettify } from "../types";
 import { addMissingVersionField } from "./addMissingVersionField";