Function queueEvents

apps/proxy/src/events/queueEvents.ts:10–112

(request: Request, env: Env)

/** 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(
        { error: 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: ApiEventLog[] = body.data.events.map((event) => {
      const timestamp = event.timestamp ?? new Date();
      return {
        ...event,
        payload: event.payload,
        timestamp,
      };
    });

    //divide updatedEvents into multiple batches of 10 (max size SQS accepts)
    const batches: ApiEventLog[][] = [];
    let currentBatch: ApiEventLog[] = [];
    for (let i = 0; i < updatedEvents.length; i++) {
      currentBatch.push(updatedEvents[i]);
      if (currentBatch.length === 10) {
        batches.push(currentBatch);
        currentBatch = [];
      }
    }
    if (currentBatch.length > 0) {
      batches.push(currentBatch);