From e32c99207dba292f9cdfaa0cae251d292d157123 Mon Sep 17 00:00:00 2001
From: Andrew Bierman <andrew@safetyamp.com>
Date: Tue, 16 Jul 2024 16:58:47 -0400
Subject: [PATCH] =?UTF-8?q?=E2=9A=A1=EF=B8=8F=20boilerplate=20to=20add=20c?=
 =?UTF-8?q?loudflare=20queues?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 server/src/index.ts          | 19 ++++++++++++-----
 server/src/queue/index.ts    | 41 ++++++++++++++++++++++++++++++++++++
 server/wrangler.toml.example | 17 +++++++++++++--
 3 files changed, 70 insertions(+), 7 deletions(-)
 create mode 100644 server/src/queue/index.ts

diff --git a/server/src/index.ts b/server/src/index.ts
index 8cf313258..60bd0f589 100644
--- a/server/src/index.ts
+++ b/server/src/index.ts
@@ -1,14 +1,16 @@
 import { Hono } from 'hono';
+import { cors } from 'hono/cors';
+import { logger } from 'hono/logger';
 import { fetchHandler } from 'trpc-playground/handlers/fetch';
 import { appRouter } from './routes/trpcRouter';
 import { honoTRPCServer } from './trpc/server';
-import { cors } from 'hono/cors';
 import { securityHeaders } from './middleware/securityHeaders';
 import { enforceHttps } from './middleware/enforceHttps';
 import router from './routes';
 import { CORS_METHODS } from './config';
+import { queue } from './queue';
 
-interface Bindings {
+export interface Bindings {
   [key: string]: any;
   DB: IDBDatabase;
   JWT_VERIFICATION_KEY: string;
@@ -48,8 +50,7 @@ app.use('*', async (c, next) => {
 });
 
 // SETUP LOGGING
-//  tRPC is already logging requests, but you can add your own middleware
-//  app.use('*', logger());
+app.use('*', logger());
 
 // SETUP TRPC SERVER
 app.use(`${TRPC_API_ENDPOINT}/*`, honoTRPCServer({ router: appRouter }));
@@ -67,4 +68,12 @@ app.use(TRPC_PLAYGROUND_ENDPOINT, async (c, next) => {
 // SET UP HTTP ROUTES
 app.route(`${HTTP_ENDPOINT}`, router);
 
-export default app;
+// SETUP CLOUDFLARE WORKER WITH EVENT HANDLERS
+const worker = {
+  ...app,
+  fetch: app.fetch,
+  queue,
+};
+
+// EXPORT WORKER
+export default worker;
diff --git a/server/src/queue/index.ts b/server/src/queue/index.ts
new file mode 100644
index 000000000..2feb5411e
--- /dev/null
+++ b/server/src/queue/index.ts
@@ -0,0 +1,41 @@
+import { MessageBatch } from '@cloudflare/workers-types';
+import { Bindings } from '..';
+
+async function handleLogQueue(batch: MessageBatch<Error>, env: Bindings) {
+  for (const message of batch.messages) {
+    console.log('Processing log message:', message.body);
+    // Add your log processing logic here
+  }
+}
+
+const handleQueueDefault = async (
+  batch: MessageBatch<Error>,
+  env: Bindings,
+) => {
+  console.error(`No handler found for queue: ${batch.queue}`);
+  // Optionally, implement a default behavior or error handling
+};
+
+// Add more handler functions as needed
+
+// Create a Map for handlers
+const queueHandlersMap = new Map<
+  string,
+  (batch: MessageBatch<Error>, env: Bindings) => Promise<void>
+>([
+  ['log-queue', handleLogQueue],
+  // Add more handlers here
+]);
+
+export async function queue(
+  batch: MessageBatch<Error>,
+  env: Bindings,
+): Promise<void> {
+  try {
+    const handler = queueHandlersMap.get(batch.queue) || handleQueueDefault;
+    await handler(batch, env);
+  } catch (error) {
+    console.error(`Error processing queue: ${batch.queue}`, error);
+    // Optionally, add more error handling logic here
+  }
+}
diff --git a/server/wrangler.toml.example b/server/wrangler.toml.example
index 90f5ed2e8..5c29c5904 100644
--- a/server/wrangler.toml.example
+++ b/server/wrangler.toml.example
@@ -1,14 +1,27 @@
 name = "packrat_api"
 main = "src/index.ts"
 compatibility_date = "2024-03-14"
-compatibility_flags = ["nodejs_compat"]
-# node_compat = true # Sometimes this is needed for tests 
+compatibility_flags = ["nodejs_compat"] # Sometimes this is needed for tests
+# node_compat = true
 
 [[ d1_databases ]]
 binding = "DB"
 database_name = "production"
 database_id = "3f9677cd-7dd9-4a2c-92c6-be5dbbf47baa"
 
+[[queues.producers]]
+  queue = "packrat-error-queue"
+  binding = "ERROR_QUEUE"
+
+[[queues.consumers]]
+  queue = "packrat-error-queue"
+  max_batch_size = 100
+  max_batch_timeout = 30
+
+[[r2_buckets]]
+  bucket_name = "packrat-error-bucket"
+  binding = "ERROR_BUCKET"
+
 # Add vars below
 # [vars]
 # GOOGLE_CLIENT_ID=""