feat: Add new feature deduplication

README.md

@@ -29,7 +29,7 @@ wrangler login
 cp wrangler.toml_example wrangler.toml
 ```
 
-- Change everywhere example.com to your domain
+- Change everywhere `example.com` to your domain
 
 
 ### Create queue
@@ -38,12 +38,19 @@ wrangler queues create cf-n8n-proxy-production -e production
 wrangler queues create dlq-cf-n8n-proxy-production -e production
 ```
 
+### Enable deduplication feature
+```
+wrangler kv:namespace create cf-n8n-proxy
+```
+- Replace `<mykvid>` with the response ID in the wrangler.toml file.
+- Change `DEDUPLICATION=false` to `DEDUPLICATION=true` in the wrangler.toml file.
+- `DEDUPLICATION_TTL` is optional; by default, it is set to 60 seconds.
+
 ### Deploy your service
 ```
 wrangler deploy -e production
 ```
 
-
 ### How to test in local?
 ```
 wrangler dev

package.json

@@ -1,19 +1,25 @@
 {
-  "name": "proxy",
+  "name": "cf-n8n-proxy",
   "scripts": {
     "dev": "wrangler dev"
   },
   "devDependencies": {
-    "@types/node": "20.11.28",
     "@cloudflare/workers-types": "4.20240314.0",
+    "@types/node": "20.11.28",
     "eslint-config-prettier": "9.1.0",
     "eslint-plugin-prettier": "5.1.3",
+    "pnpm": "latest",
     "prettier": "3.2.5",
     "typescript": "5.4.2",
     "wrangler": "3.34.2"
   },
   "dependencies": {
-    "itty-fetcher": "^0.9.4",
+    "cf-workers-hash": "0.0.3",
+    "itty-fetcher": "0.9.4",
     "itty-router": "4.2.1"
+  },
+  "engines": {
+    "node": ">=18",
+    "pnpm": ">=8"
   }
 }

src/index.ts

@@ -1,11 +1,34 @@
 import { createCors, error, Router } from 'itty-router';
+import { hasher } from 'cf-workers-hash';
 
 const { preflight, corsify } = createCors({
   methods: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS', 'HEAD'],
 });
 const router = Router();
 
-export async function proxy (request: Request, env: Env) {
+async function deduplication(request: Request, env: Env) {
+  if (!env.DEDUPLICATION) return false;
+  try {
+    const url = new URL(request.url);
+    const body = await request.text();
+    const contentLength = new TextEncoder().encode(body).length;
+    const key = await hasher(`${request.method}-${url.pathname}-${body}-${contentLength}`, 'SHA-256');
+    const ttl = env.DEDUPLICATION_TTL || 60;
+    const value = await env.DEDUPLICATION_KV.get(key);
+    if (value === null) {
+      console.debug(`Deduplication key not found: ${key}`)
+      await env.DEDUPLICATION_KV.put(key, 't', { expirationTtl: ttl });
+      return false;
+    }
+    console.debug(`Deduplication key found: ${key}`)
+    return true;
+  } catch (e) {
+    console.error(JSON.stringify(e));
+    return false;
+  }
+}
+
+async function proxy (request: Request, env: Env) {
   let url = new URL(request.url);
   const proxyUrl = `${env.PROXY_DOMAIN}${url.pathname}`;
   switch (request.method) {
@@ -18,6 +41,12 @@ export async function proxy (request: Request, env: Env) {
     case 'PUT':
     case 'PATCH':
     case 'DELETE':
+      if (await deduplication(request, env)) {
+        return new Response(null, {
+          status: 202,
+          statusText: 'Accepted',
+        });
+      }
       return await fetch(proxyUrl, {
         method: request.method,
         body: request.body,
@@ -48,7 +77,7 @@ export async function proxy (request: Request, env: Env) {
   }
 }
 
-export async function createNewRequest(request: Request): Promise<Request> {
+async function createNewRequest(request: Request): Promise<Request> {
   const { url, method, headers } = request;
   if (method === 'GET' || method === 'HEAD') {
     return new Request(url, { method, headers });
@@ -87,21 +116,21 @@ router
         }));
         return error(202, 'Request sent to error queue.')
       }
-      console.log(JSON.stringify({ status: response.status, statusText: response.statusText, headers: response.headers, body: await responseClone.text()}));
+      console.info(JSON.stringify({ status: response.status, statusText: response.statusText, headers: response.headers, body: await responseClone.text()}));
       return response;
     } catch (e) {
-      console.log(JSON.stringify(e));
+      console.error(JSON.stringify(e));
       return error(500, 'Something went wrong');
     }
   })
   .all(`/webhook-test/:id`, async (request: Request, env: Env, _: ExecutionContext) => {
     try {
       const response = await proxy(request, env);
       const responseClone = response.clone();
-      console.log(JSON.stringify({ status: response.status, statusText: response.statusText, headers: response.headers, body: await responseClone.text()}));
+      console.info(JSON.stringify({ status: response.status, statusText: response.statusText, headers: response.headers, body: await responseClone.text()}));
       return response;
     } catch (e) {
-      console.log(JSON.stringify(e));
+      console.error(JSON.stringify(e));
       return error(500, 'Something went wrong');
     }
   })
@@ -121,7 +150,7 @@ export default {
           body: json.body || null,
         });
         const response = await proxy(request, env);
-        console.log(JSON.stringify(response));
+        console.error(JSON.stringify(response));
       } catch (error) {
         batch.retryAll();
       }

worker-configuration.d.ts

@@ -8,4 +8,7 @@ interface Env {
   readonly PROXY_DOMAIN: string;
   readonly WEBHOOK_PATH: string;
   readonly WEBHOOK_TEST_PATH: string;
+  readonly DEDUPLICATION: boolean;
+  readonly DEDUPLICATION_TTL: number;
+  readonly DEDUPLICATION_KV: KVNamespace;
 }

wrangler.toml_example

@@ -14,10 +14,15 @@ queue = "cf-n8n-proxy"
 max_batch_size = 100
 max_batch_timeout = 30
 
+[[kv_namespaces]]
+binding = "DEDUPLICATION_KV"
+id = "<mykvid>"
+
 [vars]
 ENVIRONMENT = "dev"
 PROXY_DOMAIN = "https://example.com"
-
+DEDUPLICATION = false
+DEDUPLICATION_TTL = 60
 
 [[env.production.queues.producers]]
 queue = "cf-n8n-proxy-production"
@@ -30,14 +35,20 @@ max_batch_timeout = 30
 max_retries = 100
 dead_letter_queue = "dlq-cf-n8n-proxy-production"
 
+[[env.production.kv_namespaces]]
+binding = "DEDUPLICATION_KV"
+id = "<mykvid>"
+
 [env.production]
 workers_dev = false
 
 routes = [
-    { pattern = "example.com/webhook/*", zone_name = "papp.ai" },
-    { pattern = "example.com/webhook-test/*", zone_name = "papp.ai" },
+    { pattern = "example.com/webhook/*", zone_name = "example.com" },
+    { pattern = "example.com/webhook-test/*", zone_name = "example.com" },
 ]
 
 [env.production.vars]
 ENVIRONMENT = "production"
-PROXY_DOMAIN = "https://example.com"
+PROXY_DOMAIN = "https://example.com"
+DEDUPLICATION = false
+DEDUPLICATION_TTL = 60